From bd4a3baf42e22fad7b74dfa237f155bb009657c7 Mon Sep 17 00:00:00 2001 From: zhom <2717306+zhom@users.noreply.github.com> Date: Sun, 1 Nov 2026 14:04:56 +0400 Subject: [PATCH] feat: sync encryption at rest --- donut-sync/.env.example | 8 + donut-sync/Cargo.lock | 44 +++ donut-sync/Cargo.toml | 2 +- donut-sync/README.md | 4 + donut-sync/src/at_rest.rs | 440 ++++++++++++++++++++++ donut-sync/src/at_rest_tests.rs | 347 +++++++++++++++++ donut-sync/src/config.rs | 2 + donut-sync/src/http.rs | 54 ++- donut-sync/src/lib.rs | 115 ++++-- donut-sync/src/memory_store.rs | 237 ++++++++++++ donut-sync/src/migrate.rs | 638 ++++++++++++++++++++++++++++++++ donut-sync/src/s3.rs | 157 ++++++-- donut-sync/src/sealed_store.rs | 133 +++++++ donut-sync/src/store.rs | 52 +++ donut-sync/src/subscribe.rs | 1 + donut-sync/src/sync.rs | 82 +++- donut-sync/src/sync_tests.rs | 192 ++-------- donut-sync/src/time.rs | 4 + donut-sync/src/transfer.rs | 395 ++++++++++++++++++++ donut-sync/tests/at_rest.rs | 452 ++++++++++++++++++++++ donut-sync/tests/common/mod.rs | 35 +- 21 files changed, 3157 insertions(+), 237 deletions(-) create mode 100644 donut-sync/src/at_rest.rs create mode 100644 donut-sync/src/at_rest_tests.rs create mode 100644 donut-sync/src/memory_store.rs create mode 100644 donut-sync/src/migrate.rs create mode 100644 donut-sync/src/sealed_store.rs create mode 100644 donut-sync/src/transfer.rs create mode 100644 donut-sync/tests/at_rest.rs diff --git a/donut-sync/.env.example b/donut-sync/.env.example index c1702275..717c7592 100644 --- a/donut-sync/.env.example +++ b/donut-sync/.env.example @@ -33,3 +33,11 @@ S3_FORCE_PATH_STYLE=true # URL it cannot open: /health and /readyz stay green while every transfer fails. # Defaults to S3_ENDPOINT, which is correct when storage is already public. # S3_PUBLIC_ENDPOINT=https://storage.example.com + +# SYNC_AT_REST_KEY: encrypts every stored object with this server's key, even +# when E2E encryption is off in Donut Browser. Transfers then go through this +# server instead of straight to storage. Format id:base64 (32 bytes), e.g. +# k1:$(openssl rand -base64 32). To rotate, put the new key first: +# k2:...,k1:... The first key encrypts; every key listed decrypts. +# Objects stored before stay readable; `donut-sync migrate-at-rest` encrypts them. +# SYNC_AT_REST_KEY diff --git a/donut-sync/Cargo.lock b/donut-sync/Cargo.lock index 180f2f62..ed446624 100644 --- a/donut-sync/Cargo.lock +++ b/donut-sync/Cargo.lock @@ -256,6 +256,12 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" +[[package]] +name = "futures-io" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53c0fa8157de1303bfffdaa1cc2a673bfffb60102f76b0ef4441659124373fed" + [[package]] name = "futures-macro" version = "0.3.34" @@ -267,6 +273,12 @@ dependencies = [ "syn 3.0.9", ] +[[package]] +name = "futures-sink" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1944426bf7d03f1d14f708785e4b33efd750b36d48a157b836b3efc15ede8e1d" + [[package]] name = "futures-task" version = "0.3.34" @@ -280,8 +292,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" dependencies = [ "futures-core", + "futures-io", "futures-macro", + "futures-sink", "futures-task", + "memchr", "pin-project-lite", "slab", ] @@ -763,6 +778,7 @@ dependencies = [ "base64", "bytes", "futures-core", + "futures-util", "http", "http-body", "http-body-util", @@ -781,12 +797,14 @@ dependencies = [ "sync_wrapper", "tokio", "tokio-rustls", + "tokio-util", "tower", "tower-http", "tower-service", "url", "wasm-bindgen", "wasm-bindgen-futures", + "wasm-streams", "web-sys", ] @@ -1180,6 +1198,19 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-util" +version = "0.7.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e464cf451ba96ebfc6f9b6542f17ee8b8956e33f1e40d9690624e59d7a7f8a4b" +dependencies = [ + "bytes", + "futures-core", + "futures-sink", + "pin-project-lite", + "tokio", +] + [[package]] name = "tower" version = "0.5.3" @@ -1386,6 +1417,19 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "wasm-streams" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d1ec4f6517c9e11ae630e200b2b65d193279042e28edd4a2cda233e46670bbb" +dependencies = [ + "futures-util", + "js-sys", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "web-sys" version = "0.3.107" diff --git a/donut-sync/Cargo.toml b/donut-sync/Cargo.toml index 38745aab..16fd812e 100644 --- a/donut-sync/Cargo.toml +++ b/donut-sync/Cargo.toml @@ -30,7 +30,7 @@ hyper-util = { version = "0.1.21", features = ["tokio", "server", "http1", "serv md-5 = "0.11.0" rand = "0.10.3" regex-lite = "0.1.9" -reqwest = { version = "0.13.5", default-features = false, features = ["json", "rustls-no-provider"] } +reqwest = { version = "0.13.5", default-features = false, features = ["json", "rustls-no-provider", "stream"] } ring = "0.17.14" roxmltree = "0.21.2" rustls = { version = "0.23.46", default-features = false, features = ["ring", "std", "tls12", "logging"] } diff --git a/donut-sync/README.md b/donut-sync/README.md index 95ce70d0..b2de22f7 100644 --- a/donut-sync/README.md +++ b/donut-sync/README.md @@ -20,6 +20,10 @@ SYNC_TOKEN=... S3_ENDPOINT=... S3_ACCESS_KEY_ID=... S3_SECRET_ACCESS_KEY=... \ `.env.example` lists the variables. A `.env` file in the working directory is also read. +With `SYNC_AT_REST_KEY` set, the server encrypts every object it stores, and +transfers go through it instead of straight to storage. Objects stored before +stay readable; `donut-sync migrate-at-rest --help` encrypts them in place. + ## Test The end-to-end tests run the binary against the S3 server in diff --git a/donut-sync/src/at_rest.rs b/donut-sync/src/at_rest.rs new file mode 100644 index 00000000..0de2ab06 --- /dev/null +++ b/donut-sync/src/at_rest.rs @@ -0,0 +1,440 @@ +//! Encryption at rest with the server's own key (`SYNC_AT_REST_KEY`). +//! +//! An encrypted object is a header and a run of AES-256-GCM frames: +//! +//! ```text +//! "DSAR" | version (1) | key id length (1) | key id | salt (32) +//! frame 0 | frame 1 | ... | final frame +//! ``` +//! +//! Each frame holds up to `FRAME` plaintext bytes and a 16-byte tag. The +//! object key is HKDF-SHA256 of the server key and the object's random salt; +//! a frame's nonce is its index and a final flag, so frames cannot be dropped, +//! reordered or cut off at a frame boundary. Every frame authenticates the +//! header. Only the last frame may be short; an empty object is one empty +//! final frame. The object's metadata carries `MARKER` (`1:`), so a +//! HEAD tells an encrypted object from a plaintext one and gives its +//! plaintext size. + +use std::collections::VecDeque; +use std::sync::Arc; + +use base64::Engine; +use bytes::Bytes; +use futures_util::StreamExt; +use ring::aead::{AES_256_GCM, Aad, LessSafeKey, Nonce, UnboundKey}; +use ring::hkdf; + +use crate::store::{ByteStream, StoreError}; + +/// The user-metadata name (`x-amz-meta-donut-at-rest`) of an encrypted object. +pub const MARKER: &str = "donut-at-rest"; +const MAGIC: &[u8; 4] = b"DSAR"; +const VERSION: u8 = 1; +/// Plaintext bytes per frame. +pub const FRAME: usize = 64 * 1024; +const TAG: usize = 16; +const SALT: usize = 32; +const MAX_KEY_ID: usize = 32; +const OBJECT_KEY_INFO: &[u8] = b"donut-sync at-rest v1 object key"; +const URL_KEY_INFO: &[u8] = b"donut-sync at-rest v1 transfer url"; + +pub struct Key { + id: String, + secret: [u8; 32], +} + +impl Key { + pub fn id(&self) -> &str { + &self.id + } +} + +/// The server keys. The first one encrypts; all of them decrypt. +pub struct Keyring { + keys: Vec, +} + +fn key_id_valid(id: &str) -> bool { + (1..=MAX_KEY_ID).contains(&id.len()) + && id + .bytes() + .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'-' | b'_' | b'.')) +} + +impl Keyring { + /// `id:base64,id:base64`, each key 32 bytes. + pub fn parse(value: &str) -> Result { + let mut keys: Vec = Vec::new(); + for entry in value.split(',').map(str::trim).filter(|e| !e.is_empty()) { + let (id, encoded) = entry + .split_once(':') + .ok_or("SYNC_AT_REST_KEY entries are id:base64, e.g. k1:$(openssl rand -base64 32)")?; + if !key_id_valid(id) { + return Err(format!( + "SYNC_AT_REST_KEY key id '{id}' must be 1-{MAX_KEY_ID} letters, digits, '-', '_' or '.'" + )); + } + if keys.iter().any(|k| k.id == id) { + return Err(format!("SYNC_AT_REST_KEY names key id '{id}' twice")); + } + let unpadded = encoded.trim().trim_end_matches('='); + let decoded = base64::engine::general_purpose::STANDARD_NO_PAD + .decode(unpadded) + .or_else(|_| base64::engine::general_purpose::URL_SAFE_NO_PAD.decode(unpadded)) + .map_err(|_| format!("SYNC_AT_REST_KEY key '{id}' is not base64"))?; + let secret: [u8; 32] = decoded + .try_into() + .map_err(|_| format!("SYNC_AT_REST_KEY key '{id}' must be 32 bytes"))?; + keys.push(Key { + id: id.to_string(), + secret, + }); + } + if keys.is_empty() { + return Err("SYNC_AT_REST_KEY holds no key".to_string()); + } + Ok(Self { keys }) + } + + pub fn key_count(&self) -> usize { + self.keys.len() + } + + pub fn current(&self) -> &Key { + &self.keys[0] + } + + fn find(&self, id: &str) -> Option<&Key> { + self.keys.iter().find(|k| k.id == id) + } + + /// The HMAC keys that sign transfer URLs: the current key's first. + pub fn url_keys(&self) -> Vec<[u8; 32]> { + self + .keys + .iter() + .map(|key| derive(&key.secret, &[0; SALT], URL_KEY_INFO)) + .collect() + } + + /// The marker value an object sealed now carries. + pub fn marker(&self) -> String { + format!("{VERSION}:{}", self.current().id) + } +} + +struct Len(usize); + +impl hkdf::KeyType for Len { + fn len(&self) -> usize { + self.0 + } +} + +fn derive(secret: &[u8; 32], salt: &[u8], info: &[u8]) -> [u8; 32] { + let prk = hkdf::Salt::new(hkdf::HKDF_SHA256, salt).extract(secret); + let mut out = [0u8; 32]; + prk + .expand(&[info], Len(32)) + .and_then(|okm| okm.fill(&mut out)) + .expect("32 bytes is a valid HKDF-SHA256 length"); + out +} + +fn object_key(secret: &[u8; 32], salt: &[u8]) -> LessSafeKey { + let key = derive(secret, salt, OBJECT_KEY_INFO); + LessSafeKey::new(UnboundKey::new(&AES_256_GCM, &key).expect("32 bytes is an AES-256 key")) +} + +fn nonce(index: u64, last: bool) -> Nonce { + let mut bytes = [0u8; 12]; + bytes[3..11].copy_from_slice(&index.to_be_bytes()); + bytes[11] = u8::from(last); + Nonce::assume_unique_for_key(bytes) +} + +fn header_len(key_id: &str) -> u64 { + (MAGIC.len() + 2 + key_id.len() + SALT) as u64 +} + +fn frames(plain: u64) -> u64 { + plain.div_ceil(FRAME as u64).max(1) +} + +/// The stored size of `plain` bytes sealed under `key_id`. +pub fn sealed_len(key_id: &str, plain: u64) -> u64 { + header_len(key_id) + plain + frames(plain) * TAG as u64 +} + +/// The plaintext size of a sealed object of `sealed` bytes, or `None` when no +/// sealed object has that size. +pub fn plain_len(key_id: &str, sealed: u64) -> Option { + let body = sealed.checked_sub(header_len(key_id))?; + let full = (FRAME + TAG) as u64; + let (count, rest) = (body / full, body % full); + match rest { + 0 if count > 0 => Some(count * FRAME as u64), + rest if rest >= TAG as u64 => Some(count * FRAME as u64 + rest - TAG as u64), + _ => None, + } +} + +/// The key id a marker value names, for this format version. +pub fn marker_key_id(marker: &str) -> Option<&str> { + marker + .strip_prefix(&format!("{VERSION}:")) + .filter(|id| key_id_valid(id)) +} + +fn corrupt(message: &str) -> StoreError { + StoreError::Corrupt(message.to_string()) +} + +/// Encrypts a byte stream, frame by frame. +pub struct Sealer { + key: LessSafeKey, + header: Vec, + index: u64, + pending: Vec, +} + +impl Sealer { + /// A sealer and the header that starts its output. + pub fn new(keyring: &Keyring) -> (Self, Bytes) { + let current = keyring.current(); + let mut salt = [0u8; SALT]; + rand::fill(&mut salt); + let mut header = Vec::with_capacity(header_len(¤t.id) as usize); + header.extend_from_slice(MAGIC); + header.push(VERSION); + header.push(current.id.len() as u8); + header.extend_from_slice(current.id.as_bytes()); + header.extend_from_slice(&salt); + let sealer = Self { + key: object_key(¤t.secret, &salt), + header: header.clone(), + index: 0, + pending: Vec::new(), + }; + (sealer, Bytes::from(header)) + } + + fn seal(&mut self, mut frame: Vec, last: bool) -> Bytes { + self + .key + .seal_in_place_append_tag(nonce(self.index, last), Aad::from(&self.header), &mut frame) + .expect("a frame is far below the AES-GCM limit"); + self.index += 1; + Bytes::from(frame) + } + + /// Seals every full frame that is known not to be the last. + pub fn push(&mut self, data: &[u8], out: &mut VecDeque) { + self.pending.extend_from_slice(data); + while self.pending.len() > FRAME { + let rest = self.pending.split_off(FRAME); + let frame = std::mem::replace(&mut self.pending, rest); + out.push_back(self.seal(frame, false)); + } + } + + pub fn finish(mut self) -> Bytes { + let frame = std::mem::take(&mut self.pending); + self.seal(frame, true) + } +} + +/// Decrypts a sealed byte stream, frame by frame. +pub struct Opener { + keyring: Arc, + key: Option<(LessSafeKey, Vec)>, + index: u64, + pending: Vec, +} + +impl Opener { + pub fn new(keyring: Arc) -> Self { + Self { + keyring, + key: None, + index: 0, + pending: Vec::new(), + } + } + + fn read_header(&mut self) -> Result { + if self.pending.len() < MAGIC.len() + 2 { + return Ok(false); + } + if &self.pending[..MAGIC.len()] != MAGIC { + return Err(corrupt("sealed object has no header")); + } + if self.pending[MAGIC.len()] != VERSION { + return Err(corrupt("sealed object has an unknown version")); + } + let id_len = self.pending[MAGIC.len() + 1] as usize; + let total = MAGIC.len() + 2 + id_len + SALT; + if self.pending.len() < total { + return Ok(false); + } + let id = std::str::from_utf8(&self.pending[MAGIC.len() + 2..MAGIC.len() + 2 + id_len]) + .map_err(|_| corrupt("sealed object has an invalid key id"))?; + let key = self + .keyring + .find(id) + .ok_or_else(|| StoreError::Corrupt(format!("sealed object needs unknown key id '{id}'")))?; + let key = object_key(&key.secret, &self.pending[total - SALT..total]); + let rest = self.pending.split_off(total); + let header = std::mem::replace(&mut self.pending, rest); + self.key = Some((key, header)); + Ok(true) + } + + fn open(&mut self, mut frame: Vec, last: bool) -> Result { + let (key, header) = self.key.as_ref().expect("the header is read first"); + let plain = key + .open_in_place(nonce(self.index, last), Aad::from(header), &mut frame) + .map_err(|_| corrupt("sealed object failed authentication"))? + .len(); + frame.truncate(plain); + self.index += 1; + Ok(Bytes::from(frame)) + } + + pub fn push(&mut self, data: &[u8], out: &mut VecDeque) -> Result<(), StoreError> { + self.pending.extend_from_slice(data); + if self.key.is_none() && !self.read_header()? { + return Ok(()); + } + while self.pending.len() > FRAME + TAG { + let rest = self.pending.split_off(FRAME + TAG); + let frame = std::mem::replace(&mut self.pending, rest); + out.push_back(self.open(frame, false)?); + } + Ok(()) + } + + pub fn finish(mut self) -> Result { + if self.key.is_none() && !self.read_header()? { + return Err(corrupt("sealed object is cut off in its header")); + } + if self.pending.len() < TAG { + return Err(corrupt("sealed object is cut off")); + } + let frame = std::mem::take(&mut self.pending); + self.open(frame, true) + } +} + +enum Codec { + Seal(Option, Option), + Open(Option), +} + +impl Codec { + fn push(&mut self, data: &[u8], out: &mut VecDeque) -> Result<(), StoreError> { + match self { + Codec::Seal(sealer, header) => { + if let Some(header) = header.take() { + out.push_back(header); + } + sealer.as_mut().expect("not finished").push(data, out); + Ok(()) + } + Codec::Open(opener) => opener.as_mut().expect("not finished").push(data, out), + } + } + + fn finish(&mut self, out: &mut VecDeque) -> Result<(), StoreError> { + match self { + Codec::Seal(sealer, header) => { + if let Some(header) = header.take() { + out.push_back(header); + } + out.push_back(sealer.take().expect("finished once").finish()); + Ok(()) + } + Codec::Open(opener) => { + out.push_back(opener.take().expect("finished once").finish()?); + Ok(()) + } + } + } +} + +fn transform(input: ByteStream, codec: Codec) -> ByteStream { + struct State { + input: ByteStream, + codec: Codec, + out: VecDeque, + done: bool, + } + let state = State { + input, + codec, + out: VecDeque::new(), + done: false, + }; + Box::pin(futures_util::stream::unfold( + state, + |mut state| async move { + loop { + if let Some(chunk) = state.out.pop_front() { + if chunk.is_empty() { + continue; + } + return Some((Ok(chunk), state)); + } + if state.done { + return None; + } + let step = match state.input.next().await { + Some(Ok(data)) => state.codec.push(&data, &mut state.out), + Some(Err(e)) => Err(e), + None => { + state.done = true; + state.codec.finish(&mut state.out) + } + }; + if let Err(e) = step { + state.done = true; + state.out.clear(); + return Some((Err(e), state)); + } + } + }, + )) +} + +/// `input` sealed under the current key. +pub fn seal_stream(keyring: &Keyring, input: ByteStream) -> ByteStream { + let (sealer, header) = Sealer::new(keyring); + transform(input, Codec::Seal(Some(sealer), Some(header))) +} + +/// `input`, a sealed object, opened. A forged, cut or wrongly keyed object +/// ends the stream with `StoreError::Corrupt`. +pub fn open_stream(keyring: Arc, input: ByteStream) -> ByteStream { + transform(input, Codec::Open(Some(Opener::new(keyring)))) +} + +pub fn seal(keyring: &Keyring, plain: &[u8]) -> Vec { + let (mut sealer, header) = Sealer::new(keyring); + let mut out = VecDeque::from([header]); + sealer.push(plain, &mut out); + out.push_back(sealer.finish()); + out.into_iter().flat_map(|b| b.to_vec()).collect() +} + +pub fn open(keyring: Arc, sealed: &[u8]) -> Result, StoreError> { + let mut opener = Opener::new(keyring); + let mut out = VecDeque::new(); + opener.push(sealed, &mut out)?; + out.push_back(opener.finish()?); + Ok(out.into_iter().flat_map(|b| b.to_vec()).collect()) +} + +#[cfg(test)] +#[path = "at_rest_tests.rs"] +mod tests; diff --git a/donut-sync/src/at_rest_tests.rs b/donut-sync/src/at_rest_tests.rs new file mode 100644 index 00000000..ba7bf2a1 --- /dev/null +++ b/donut-sync/src/at_rest_tests.rs @@ -0,0 +1,347 @@ +use super::*; +use crate::store::collect; + +const K1: &str = "k1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="; +const K2: &str = "k2:AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE="; + +fn keyring(value: &str) -> Arc { + Arc::new(Keyring::parse(value).unwrap()) +} + +fn sample(len: usize) -> Vec { + (0..len).map(|i| (i * 31 + 7) as u8).collect() +} + +const SIZES: [usize; 9] = [ + 0, + 1, + FRAME - 1, + FRAME, + FRAME + 1, + 2 * FRAME - 1, + 2 * FRAME, + 2 * FRAME + 1, + 3 * FRAME + 1234, +]; + +fn header_bytes(id: &str) -> usize { + header_len(id) as usize +} + +#[test] +fn every_boundary_size_round_trips_with_the_size_math() { + let keys = keyring(K1); + for size in SIZES { + let plain = sample(size); + let sealed = seal(&keys, &plain); + assert_eq!( + sealed.len() as u64, + sealed_len("k1", size as u64), + "size {size}" + ); + assert_eq!( + plain_len("k1", sealed.len() as u64), + Some(size as u64), + "size {size}" + ); + assert_eq!(open(keys.clone(), &sealed).unwrap(), plain, "size {size}"); + } +} + +#[test] +fn an_empty_object_is_one_empty_final_frame() { + let keys = keyring(K1); + let sealed = seal(&keys, b""); + assert_eq!(sealed.len(), header_bytes("k1") + TAG); + assert_eq!(open(keys, &sealed).unwrap(), b""); +} + +#[test] +fn sizes_no_sealed_object_has_are_refused() { + let header = header_len("k1"); + assert_eq!(plain_len("k1", 0), None); + assert_eq!(plain_len("k1", header), None); + assert_eq!(plain_len("k1", header + TAG as u64 - 1), None); + assert_eq!(plain_len("k1", header + (FRAME + TAG) as u64 + 3), None); + assert_eq!(plain_len("k1", header + TAG as u64), Some(0)); +} + +#[test] +fn any_flipped_byte_is_refused() { + let keys = keyring(K1); + let sealed = seal(&keys, &sample(FRAME + 100)); + let header = header_bytes("k1"); + // Magic, version, key id, salt, first frame, first tag, last frame, last tag. + for at in [ + 0, + 4, + 6, + header - 1, + header, + header + FRAME + TAG - 1, + header + FRAME + TAG, + sealed.len() - 1, + ] { + let mut forged = sealed.clone(); + forged[at] ^= 1; + assert!( + matches!(open(keys.clone(), &forged), Err(StoreError::Corrupt(_))), + "byte {at}" + ); + } +} + +#[test] +fn truncation_anywhere_is_refused() { + let keys = keyring(K1); + let sealed = seal(&keys, &sample(2 * FRAME + 10)); + let header = header_bytes("k1"); + let frame = FRAME + TAG; + for cut in [ + 0, + 3, + header - 1, + header, + header + 5, + header + frame, + header + 2 * frame, + sealed.len() - 1, + ] { + assert!( + matches!( + open(keys.clone(), &sealed[..cut]), + Err(StoreError::Corrupt(_)) + ), + "cut at {cut}" + ); + } + // Exactly at a frame boundary of an object whose last frame is full. + let full = seal(&keys, &sample(2 * FRAME)); + assert!(open(keys.clone(), &full[..header + frame]).is_err()); +} + +#[test] +fn reordered_dropped_or_appended_frames_are_refused() { + let keys = keyring(K1); + let sealed = seal(&keys, &sample(3 * FRAME)); + let header = header_bytes("k1"); + let frame = FRAME + TAG; + let part = |i: usize| &sealed[header + i * frame..header + (i + 1) * frame]; + let swapped = [&sealed[..header], part(1), part(0), part(2)].concat(); + assert!(open(keys.clone(), &swapped).is_err()); + let dropped = [&sealed[..header], part(0), part(2)].concat(); + assert!(open(keys.clone(), &dropped).is_err()); + let appended = [&sealed[..], &[0u8; TAG]].concat(); + assert!(open(keys.clone(), &appended).is_err()); +} + +#[test] +fn a_frame_from_another_object_is_refused() { + let keys = keyring(K1); + let a = seal(&keys, &sample(10)); + let b = seal(&keys, &sample(10)); + let header = header_bytes("k1"); + let spliced = [&a[..header], &b[header..]].concat(); + assert!(open(keys, &spliced).is_err()); +} + +#[test] +fn the_wrong_key_or_an_unknown_key_id_is_refused() { + let sealed = seal(&keyring(K1), b"secret"); + let impostor = keyring("k1:AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE="); + assert!(matches!( + open(impostor, &sealed), + Err(StoreError::Corrupt(_)) + )); + let other = keyring(K2); + match open(other, &sealed) { + Err(StoreError::Corrupt(message)) => assert!(message.contains("'k1'")), + other => panic!("{other:?}"), + } +} + +#[test] +fn an_old_key_still_reads_after_rotation_and_new_writes_use_the_first() { + let old = seal(&keyring(K1), b"before"); + let rotated = keyring(&format!("{K2},{K1}")); + assert_eq!(open(rotated.clone(), &old).unwrap(), b"before"); + let new = seal(&rotated, b"after"); + assert_eq!(&new[6..8], b"k2"); + assert_eq!(rotated.marker(), "1:k2"); + assert!(open(keyring(K1), &new).is_err()); +} + +#[test] +fn the_same_plaintext_seals_differently_each_time() { + let keys = keyring(K1); + assert_ne!(seal(&keys, b"same"), seal(&keys, b"same")); +} + +fn chunked(data: Vec, sizes: &[usize]) -> crate::store::ByteStream { + let mut chunks = Vec::new(); + let mut at = 0; + let mut i = 0; + while at < data.len() { + let n = sizes[i % sizes.len()].min(data.len() - at); + chunks.push(Ok(Bytes::copy_from_slice(&data[at..at + n]))); + at += n; + i += 1; + } + Box::pin(futures_util::stream::iter(chunks)) +} + +#[tokio::test] +async fn streams_match_the_whole_buffer_functions_for_any_chunking() { + let keys = keyring(K1); + for size in SIZES { + let plain = sample(size); + for sizes in [ + &[1usize][..], + &[7, 4093, 65_537], + &[FRAME], + &[3 * FRAME + 9999], + ] { + if size > 3 * FRAME && sizes == [1] { + continue; + } + let sealed = collect(seal_stream(&keys, chunked(plain.clone(), sizes))) + .await + .unwrap(); + assert_eq!(sealed.len() as u64, sealed_len("k1", size as u64)); + assert_eq!(open(keys.clone(), &sealed).unwrap(), plain); + let opened = collect(open_stream(keys.clone(), chunked(sealed, sizes))) + .await + .unwrap(); + assert_eq!(opened, plain, "size {size} chunks {sizes:?}"); + } + } +} + +#[tokio::test] +async fn a_cut_stream_ends_in_an_error_not_a_short_body() { + let keys = keyring(K1); + let sealed = seal(&keys, &sample(2 * FRAME + 5)); + let cut = sealed[..sealed.len() - 3].to_vec(); + let result = collect(open_stream(keys, chunked(cut, &[1000]))).await; + assert!(matches!(result, Err(StoreError::Corrupt(_)))); +} + +#[test] +fn keyrings_parse_strictly() { + assert_eq!(keyring(&format!(" {K2} , {K1} ")).current().id(), "k2"); + let url_safe = keyring("a.b_c-1:AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE"); + assert_eq!(url_safe.key_count(), 1); + for bad in [ + "", + ",", + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=", + "k1:AAAA", + "k1:not base64!", + "k 1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=", + ":AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=", + &format!("{K1},{K1}"), + &format!( + "{}:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=", + "k".repeat(33) + ), + ] { + assert!(Keyring::parse(bad).is_err(), "{bad}"); + } +} + +#[test] +fn markers_name_their_key() { + assert_eq!(marker_key_id("1:k1"), Some("k1")); + assert_eq!(marker_key_id("2:k1"), None); + assert_eq!(marker_key_id("1:"), None); +} + +mod sealed_store { + use super::*; + use crate::memory_store::MemoryStore; + use crate::sealed_store::SealedStore; + use crate::store::{Store, Upload, once}; + + fn stores() -> (Arc, SealedStore) { + let raw = Arc::new(MemoryStore::default()); + let sealed = SealedStore::new(raw.clone(), keyring(K1)); + (raw, sealed) + } + + #[tokio::test] + async fn writes_are_sealed_and_read_back_as_written() { + let (raw, store) = stores(); + let metadata = vec![("updated-at".to_string(), "42".to_string())]; + let plain = sample(FRAME + 3); + store + .put_object_stream( + "a", + once(plain.clone()), + Upload { + length: plain.len() as u64, + content_type: Some("application/octet-stream"), + metadata: &metadata, + if_match: None, + }, + ) + .await + .unwrap(); + let stored = raw.raw("a").unwrap(); + assert_ne!(stored.body, plain); + assert_eq!(&stored.body[..4], b"DSAR"); + assert!( + stored + .metadata + .contains(&(MARKER.to_string(), "1:k1".to_string())) + ); + + let head = store.head_object("a").await.unwrap(); + assert_eq!(head.content_length, Some(plain.len() as u64)); + assert_eq!(head.metadata, metadata); + let download = store.get_object("a").await.unwrap(); + assert_eq!( + download.head.content_type.as_deref(), + Some("application/octet-stream") + ); + assert_eq!(collect(download.body).await.unwrap(), plain); + + store + .put_object("t", b"{\"x\":1}".to_vec(), "application/json", &[]) + .await + .unwrap(); + assert_ne!(raw.get("t").unwrap(), "{\"x\":1}"); + assert_eq!(store.get_object_text("t").await.unwrap(), "{\"x\":1}"); + } + + #[tokio::test] + async fn plaintext_objects_from_before_read_unchanged() { + let (raw, store) = stores(); + raw.insert( + "old", + b"DSAR looks sealed but has no marker".to_vec(), + vec![("updated-at".to_string(), "7".to_string())], + None, + ); + let head = store.head_object("old").await.unwrap(); + assert_eq!(head.content_length, Some(35)); + assert_eq!( + head.metadata, + vec![("updated-at".to_string(), "7".to_string())] + ); + assert_eq!( + store.get_object_text("old").await.unwrap(), + "DSAR looks sealed but has no marker" + ); + } + + #[tokio::test] + async fn copies_stay_readable() { + let (_, store) = stores(); + store + .put_object("a", b"x".to_vec(), "text/plain", &[]) + .await + .unwrap(); + store.copy_object("a", "b").await.unwrap(); + assert_eq!(store.get_object_text("b").await.unwrap(), "x"); + } +} diff --git a/donut-sync/src/config.rs b/donut-sync/src/config.rs index b1269914..54f0bfac 100644 --- a/donut-sync/src/config.rs +++ b/donut-sync/src/config.rs @@ -17,6 +17,7 @@ pub struct Config { pub s3_force_path_style: Option, pub s3_public_endpoint: Option, pub port: Option, + pub at_rest_key: Option, } fn env(name: &str) -> Option { @@ -41,6 +42,7 @@ impl Config { s3_force_path_style: env("S3_FORCE_PATH_STYLE"), s3_public_endpoint: env("S3_PUBLIC_ENDPOINT"), port: env("PORT"), + at_rest_key: env("SYNC_AT_REST_KEY"), } } diff --git a/donut-sync/src/http.rs b/donut-sync/src/http.rs index 6632e4b2..ff713bc5 100644 --- a/donut-sync/src/http.rs +++ b/donut-sync/src/http.rs @@ -6,7 +6,7 @@ use std::sync::Arc; use std::time::Duration; use bytes::Bytes; -use http_body_util::combinators::BoxBody; +use http_body_util::combinators::UnsyncBoxBody; use http_body_util::{BodyExt, Full, StreamBody}; use hyper::body::{Frame, Incoming}; use hyper::http::request::Parts; @@ -19,7 +19,8 @@ use crate::jsval::Body; use crate::log; use crate::sync::SyncService; -pub type ResBody = BoxBody; +pub type BoxError = Box; +pub type ResBody = UnsyncBoxBody; /// The JSON body limit (`50mb` in body-parser terms). pub const BODY_LIMIT: usize = 50 * 1024 * 1024; @@ -32,10 +33,10 @@ pub struct AppState { pub account_url: Option, } -fn full(bytes: impl Into) -> ResBody { +pub fn full(bytes: impl Into) -> ResBody { Full::new(bytes.into()) .map_err(|never| match never {}) - .boxed() + .boxed_unsync() } fn json_response(status: u16, body: &Value) -> Response { @@ -71,7 +72,7 @@ fn header_value(parts: &Parts, name: impl header::AsHeaderName) -> Option, ) -> Result, Infallible> { let (parts, body) = request.into_parts(); + // A transfer's body is the object, streamed: it must not be read as JSON. + if let Some(signer) = state.sync.transfer() + && route_path(parts.uri.path()) == crate::transfer::PATH + { + let response = crate::transfer::serve(state.sync.store(), signer, &parts, body).await; + return Ok(with_cors(response)); + } let body = match read_json_body(&parts, body).await { Ok(body) => body, Err(response) => return Ok(*response), @@ -253,7 +261,7 @@ async fn route(state: &Arc, parts: &Parts, body: Body) -> Response ctx, Err(response) => return *response, }; - match object_route_call(&state.sync, name, &body, &ctx).await { + match object_route_call(&state.sync, name, &body, &ctx, &origin(state, parts)).await { Ok(value) => json_response(200, &value), Err(error) => error_response(error), } @@ -277,20 +285,42 @@ const OBJECT_ROUTES: [&str; 12] = [ "snapshots/restore-files", ]; +/// How devices reach this server, for the transfer URLs of a sealed store: +/// `SYNC_PUBLIC_URL`, or the address this request came to. +fn origin(state: &AppState, parts: &Parts) -> String { + if let Some(url) = &state.account_url { + return url.clone(); + } + let first = |name: &str| { + header_value(parts, name) + .and_then(|v| v.split(',').next().map(|s| s.trim().to_string())) + .filter(|v| !v.is_empty()) + }; + let scheme = match first("x-forwarded-proto").as_deref() { + Some("https") => "https", + _ => "http", + }; + let host = first("x-forwarded-host") + .or_else(|| first("host")) + .unwrap_or_else(|| "localhost".to_string()); + format!("{scheme}://{host}") +} + async fn object_route_call( sync: &SyncService, name: &str, body: &Body, ctx: &UserContext, + origin: &str, ) -> Result { match name { "stat" => sync.stat(body, ctx).await, - "presign-upload" => sync.presign_upload(body, ctx).await, - "presign-download" => sync.presign_download(body, ctx).await, + "presign-upload" => sync.presign_upload(body, ctx, origin).await, + "presign-download" => sync.presign_download(body, ctx, origin).await, "delete" => sync.delete(body, ctx).await, "list" => sync.list(body, ctx).await, - "presign-upload-batch" => sync.presign_upload_batch(body, ctx).await, - "presign-download-batch" => sync.presign_download_batch(body, ctx).await, + "presign-upload-batch" => sync.presign_upload_batch(body, ctx, origin).await, + "presign-download-batch" => sync.presign_download_batch(body, ctx, origin).await, "delete-prefix" => sync.delete_prefix(body, ctx).await, "snapshots/list" => sync.list_snapshots(body, ctx).await, "snapshots/create" => sync.create_snapshot(body, ctx).await, @@ -356,7 +386,7 @@ fn subscribe(state: &AppState, ctx: UserContext) -> Response { let stream = futures_util::stream::unfold(rx, |mut rx| async move { rx.recv() .await - .map(|bytes| (Ok::<_, Infallible>(Frame::data(bytes)), rx)) + .map(|bytes| (Ok::<_, BoxError>(Frame::data(bytes)), rx)) }); Response::builder() .header(header::CONTENT_TYPE, "text/event-stream") @@ -368,7 +398,7 @@ fn subscribe(state: &AppState, ctx: UserContext) -> Response { .header(header::PRAGMA, "no-cache") .header(header::EXPIRES, "0") .header("x-accel-buffering", "no") - .body(StreamBody::new(stream).boxed()) + .body(StreamBody::new(stream).boxed_unsync()) .expect("static response parts are valid") } diff --git a/donut-sync/src/lib.rs b/donut-sync/src/lib.rs index 9c00d77b..c1777d03 100644 --- a/donut-sync/src/lib.rs +++ b/donut-sync/src/lib.rs @@ -3,6 +3,7 @@ //! presigned URLs, so profile files go straight between devices and storage. pub mod accounts; +pub mod at_rest; pub mod auth; pub mod backend; pub mod config; @@ -11,13 +12,18 @@ pub mod http; pub mod jsval; pub mod jwt; pub mod log; +#[cfg(test)] +mod memory_store; +pub mod migrate; pub mod net; pub mod s3; +pub mod sealed_store; pub mod sigv4; pub mod store; pub mod subscribe; pub mod sync; pub mod time; +pub mod transfer; use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr}; use std::sync::Arc; @@ -28,13 +34,17 @@ use hyper_util::rt::{TokioIo, TokioTimer}; use tokio::net::TcpListener; use crate::accounts::{DonutAccounts, normalize_sync_server_url}; +use crate::at_rest::Keyring; use crate::auth::Guard; use crate::backend::Backend; use crate::config::Config; use crate::http::AppState; use crate::s3::{Location, Presigner, S3Store}; +use crate::sealed_store::SealedStore; use crate::sigv4::Credentials; +use crate::store::Store; use crate::sync::{SyncOptions, SyncService}; +use crate::transfer::TransferSigner; const DEFAULT_PORT: u16 = 3929; /// Shipped placeholders that must never guard a real server. @@ -74,6 +84,9 @@ pub fn validate_env(config: &Config) -> Result<(), String> { .to_string(), ); } + if let Some(value) = &config.at_rest_key { + Keyring::parse(value)?; + } Ok(()) } @@ -83,49 +96,95 @@ fn require(value: &Option, name: &str) -> Result { .ok_or_else(|| format!("Required environment variable {name} is not set")) } +struct Storage { + endpoint: String, + bucket: String, + region: String, + credentials: Credentials, + force_path_style: bool, +} + +impl Storage { + // No defaults for storage: a misconfigured server must not start pointed at + // an unintended or public-default backend. + fn from_config(config: &Config) -> Result { + Ok(Self { + endpoint: require(&config.s3_endpoint, "S3_ENDPOINT")?, + bucket: require(&config.s3_bucket, "S3_BUCKET")?, + region: config + .s3_region + .clone() + .unwrap_or_else(|| "us-east-1".to_string()), + credentials: Credentials { + access_key_id: require(&config.s3_access_key_id, "S3_ACCESS_KEY_ID")?, + secret_access_key: require(&config.s3_secret_access_key, "S3_SECRET_ACCESS_KEY")?, + }, + force_path_style: config.s3_force_path_style.as_deref() != Some("false"), + }) + } + + fn store(&self) -> Result { + Ok(S3Store::new( + Location::parse(&self.endpoint, &self.bucket, self.force_path_style)?, + self.credentials.clone(), + self.region.clone(), + )) + } +} + +/// The bucket itself, without sealing: for the migration. +pub fn raw_store(config: &Config) -> Result { + Storage::from_config(config)?.store() +} + /// Everything the server needs, connected and checked: storage reachable and /// the bucket created if it was missing. pub async fn build_state(config: &Config) -> Result, String> { - // No defaults for storage: a misconfigured server must not start pointed at - // an unintended or public-default backend. - let endpoint = require(&config.s3_endpoint, "S3_ENDPOINT")?; - let region = config - .s3_region - .clone() - .unwrap_or_else(|| "us-east-1".to_string()); - let access_key_id = require(&config.s3_access_key_id, "S3_ACCESS_KEY_ID")?; - let secret_access_key = require(&config.s3_secret_access_key, "S3_SECRET_ACCESS_KEY")?; - let force_path_style = config.s3_force_path_style.as_deref() != Some("false"); - let bucket = require(&config.s3_bucket, "S3_BUCKET")?; - let credentials = Credentials { - access_key_id, - secret_access_key, - }; - + let storage = Storage::from_config(config)?; // Presigned URLs go to a desktop on another machine, so they must name a // host that machine can reach. Signing is bound to the host, so they are // signed against the public address rather than rewritten after signing. let public_endpoint = config .s3_public_endpoint .clone() - .unwrap_or_else(|| endpoint.clone()); - let store = S3Store::new( - Location::parse(&endpoint, &bucket, force_path_style)?, - credentials.clone(), - region.clone(), - ); + .unwrap_or_else(|| storage.endpoint.clone()); + let s3: Arc = Arc::new(storage.store()?); + let keyring = config + .at_rest_key + .as_deref() + .map(Keyring::parse) + .transpose()? + .map(Arc::new); + let (store, transfer): (Arc, _) = match &keyring { + Some(keyring) => { + crate::log::log( + "SyncService", + &format!( + "Encryption at rest on key={} keys={}", + keyring.current().id(), + keyring.key_count() + ), + ); + ( + Arc::new(SealedStore::new(s3, keyring.clone())), + Some(TransferSigner::new(keyring)), + ) + } + None => (s3, None), + }; let presigner = Presigner::new( - Location::parse(&public_endpoint, &bucket, force_path_style)?, - credentials, - region, + Location::parse(&public_endpoint, &storage.bucket, storage.force_path_style)?, + storage.credentials, + storage.region, ); let backend = config .backend() .map(|(url, key)| Arc::new(Backend::new(url, key))); let sync = Arc::new(SyncService::new( - Arc::new(store), + store, SyncOptions { + transfer, presigner, public_endpoint, public_endpoint_configured: config.s3_public_endpoint.is_some(), @@ -240,6 +299,10 @@ pub async fn run() -> i32 { // A `.env` in the working directory fills in what the environment lacks. let _ = dotenvy::dotenv(); let config = Config::from_env(); + let args: Vec = std::env::args().skip(1).collect(); + if args.first().map(String::as_str) == Some(migrate::COMMAND) { + return migrate::run(&config, &args[1..]).await; + } if let Err(message) = validate_env(&config) { eprintln!("{message}"); return 1; diff --git a/donut-sync/src/memory_store.rs b/donut-sync/src/memory_store.rs new file mode 100644 index 00000000..817e3c96 --- /dev/null +++ b/donut-sync/src/memory_store.rs @@ -0,0 +1,237 @@ +//! A bucket in memory, for unit tests. Hooks stand in for HEAD and LIST where +//! a test needs answers the bucket itself would not give. + +use std::collections::BTreeMap; +use std::sync::Mutex; +use std::sync::atomic::{AtomicUsize, Ordering}; + +use async_trait::async_trait; +use md5::{Digest, Md5}; + +use crate::store::{ + ByteStream, Download, HeadOutput, ListInput, ListObject, ListOutput, Store, StoreError, Upload, + collect, once, +}; + +pub type HeadHook = Box Result + Send + Sync>; +pub type ListHook = Box ListOutput + Send + Sync>; + +#[derive(Clone)] +pub struct Object { + pub body: Vec, + pub metadata: Vec<(String, String)>, + pub content_type: Option, +} + +#[derive(Default)] +pub struct MemoryStore { + pub objects: Mutex>, + pub calls: AtomicUsize, + pub head_hook: Option, + pub list_hook: Option, +} + +fn etag(body: &[u8]) -> String { + format!("\"{}\"", hex::encode(Md5::digest(body))) +} + +impl MemoryStore { + fn call(&self) { + self.calls.fetch_add(1, Ordering::SeqCst); + } + + pub fn set(&self, key: &str, body: &str) { + self.insert(key, body.as_bytes().to_vec(), Vec::new(), None); + } + + pub fn insert( + &self, + key: &str, + body: Vec, + metadata: Vec<(String, String)>, + content_type: Option, + ) { + self.objects.lock().unwrap().insert( + key.to_string(), + Object { + body, + metadata, + content_type, + }, + ); + } + + pub fn get(&self, key: &str) -> Option { + self + .raw(key) + .map(|o| String::from_utf8_lossy(&o.body).into_owned()) + } + + pub fn raw(&self, key: &str) -> Option { + self.objects.lock().unwrap().get(key).cloned() + } + + pub fn keys(&self) -> Vec { + self.objects.lock().unwrap().keys().cloned().collect() + } + + fn head(&self, key: &str) -> Result { + let object = self.raw(key).ok_or(StoreError::NotFound)?; + let mut metadata = object.metadata.clone(); + metadata.sort(); + Ok(HeadOutput { + last_modified_ms: Some(0), + content_length: Some(object.body.len() as u64), + metadata, + etag: Some(etag(&object.body)), + content_type: object.content_type, + }) + } +} + +#[async_trait] +impl Store for MemoryStore { + async fn head_bucket(&self) -> Result<(), StoreError> { + self.call(); + Ok(()) + } + + async fn create_bucket(&self) -> Result<(), StoreError> { + self.call(); + Ok(()) + } + + async fn head_object(&self, key: &str) -> Result { + self.call(); + if let Some(hook) = &self.head_hook { + return hook(key); + } + self.head(key) + } + + async fn get_object_text(&self, key: &str) -> Result { + self.call(); + self.get(key).ok_or(StoreError::NotFound) + } + + async fn put_object( + &self, + key: &str, + body: Vec, + content_type: &str, + metadata: &[(String, String)], + ) -> Result<(), StoreError> { + self.call(); + self.insert(key, body, metadata.to_vec(), Some(content_type.to_string())); + Ok(()) + } + + async fn delete_object(&self, key: &str) -> Result<(), StoreError> { + self.call(); + self.objects.lock().unwrap().remove(key); + Ok(()) + } + + async fn delete_objects(&self, keys: &[String]) -> Result<(), StoreError> { + self.call(); + let mut objects = self.objects.lock().unwrap(); + for key in keys { + objects.remove(key); + } + Ok(()) + } + + async fn copy_object(&self, from: &str, to: &str) -> Result<(), StoreError> { + self.call(); + let mut objects = self.objects.lock().unwrap(); + let object = objects.get(from).cloned().ok_or(StoreError::NotFound)?; + objects.insert(to.to_string(), object); + Ok(()) + } + + async fn list_objects_v2(&self, input: &ListInput) -> Result { + self.call(); + if let Some(hook) = &self.list_hook { + return Ok(hook(input)); + } + let prefix = input.prefix.clone().unwrap_or_default(); + let max_keys: usize = input + .max_keys + .as_deref() + .and_then(|m| m.parse().ok()) + .unwrap_or(1000); + let after = input + .continuation_token + .clone() + .or_else(|| input.start_after.clone()); + let objects = self.objects.lock().unwrap(); + let mut out = ListOutput::default(); + for (key, object) in objects.range(prefix.clone()..) { + if !key.starts_with(&prefix) { + break; + } + if after.as_ref().is_some_and(|after| key <= after) { + continue; + } + if let Some(delimiter) = &input.delimiter + && let Some(at) = key[prefix.len()..].find(delimiter.as_str()) + { + let common = key[..prefix.len() + at + delimiter.len()].to_string(); + if !out.common_prefixes.contains(&common) { + out.common_prefixes.push(common); + } + continue; + } + if out.contents.len() == max_keys { + out.is_truncated = true; + out.next_continuation_token = out.contents.last().map(|o| o.key.clone()); + break; + } + out.contents.push(ListObject { + key: key.clone(), + last_modified_ms: Some(0), + size: Some(object.body.len() as u64), + }); + } + Ok(out) + } + + async fn get_object(&self, key: &str) -> Result { + self.call(); + let head = self.head(key)?; + let body = self.raw(key).ok_or(StoreError::NotFound)?.body; + Ok(Download { + head, + body: once(body), + }) + } + + async fn put_object_stream( + &self, + key: &str, + body: ByteStream, + upload: Upload<'_>, + ) -> Result<(), StoreError> { + self.call(); + let body = collect(body).await?; + if body.len() as u64 != upload.length { + return Err(StoreError::Transport("length mismatch".to_string())); + } + if let Some(expected) = upload.if_match + && self.raw(key).map(|o| etag(&o.body)).as_deref() != Some(expected) + { + return Err(StoreError::Service { + status: 412, + code: "PreconditionFailed".to_string(), + message: String::new(), + }); + } + self.insert( + key, + body, + upload.metadata.to_vec(), + upload.content_type.map(str::to_string), + ); + Ok(()) + } +} diff --git a/donut-sync/src/migrate.rs b/donut-sync/src/migrate.rs new file mode 100644 index 00000000..01c55a79 --- /dev/null +++ b/donut-sync/src/migrate.rs @@ -0,0 +1,638 @@ +//! `donut-sync migrate-at-rest`: seals the plaintext objects already in the +//! bucket, and re-seals objects under an older key, while the server runs. +//! +//! Run it after the server runs with `SYNC_AT_REST_KEY` and the S3 URLs it +//! handed out before have expired (one hour), so no device writes plaintext +//! behind it. A rerun skips what is sealed under the current key. +//! +//! A device can write an object while it is re-sealed. Where the store honors +//! `If-Match` on PUT, the write is conditional on the ETag read. Elsewhere the +//! object is sealed into a temporary key outside every namespace, its ETag is +//! checked again, and a server-side copy puts it in place, so the window is +//! one HEAD and one copy, whatever the object's size. + +use std::path::PathBuf; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use futures_util::StreamExt; + +use crate::at_rest::{self, Keyring, MARKER}; +use crate::config::Config; +use crate::log; +use crate::sealed_store::sealed_by; +use crate::store::{ByteStream, ListInput, Store, StoreError, Upload}; + +pub const COMMAND: &str = "migrate-at-rest"; +const CONTEXT: &str = "Migrate"; +/// Temporary sealed copies. No client prefix matches it. +pub const TMP_PREFIX: &str = ".donut-sync-at-rest-tmp/"; +const PROGRESS_EVERY: Duration = Duration::from_secs(10); +/// Tries per object: storage has transient failures. +const ATTEMPTS: u32 = 3; + +const USAGE: &str = "Usage: donut-sync migrate-at-rest [--dry-run] [--prefix P] [--concurrency N] [--max-mib-per-sec N] [--checkpoint FILE] + +Seals every plaintext object in S3_BUCKET with the first key of SYNC_AT_REST_KEY, +and re-seals objects under its other keys. Safe to stop and run again. + + --dry-run Count what would change; write nothing. + --prefix P Only keys under P. + --concurrency N Objects at a time (default 4). + --max-mib-per-sec N Read at most N MiB/s (default unlimited). + --checkpoint FILE Resume after the key in FILE, and keep it current."; + +#[derive(Debug, Default, Clone)] +pub struct Options { + pub dry_run: bool, + pub prefix: Option, + pub concurrency: usize, + pub bytes_per_sec: Option, + pub checkpoint: Option, +} + +fn parse(args: &[String]) -> Result { + let mut options = Options { + concurrency: 4, + ..Options::default() + }; + let mut args = args.iter(); + while let Some(arg) = args.next() { + let mut value = |name: &str| { + args + .next() + .cloned() + .ok_or_else(|| format!("{name} needs a value")) + }; + match arg.as_str() { + "--dry-run" => options.dry_run = true, + "--prefix" => options.prefix = Some(value(arg)?), + "--concurrency" => { + options.concurrency = value(arg)? + .parse() + .ok() + .filter(|n| (1..=64).contains(n)) + .ok_or("--concurrency must be 1-64")?; + } + "--max-mib-per-sec" => { + let mib: u64 = value(arg)? + .parse() + .ok() + .filter(|n| *n > 0) + .ok_or("--max-mib-per-sec must be a positive whole number")?; + options.bytes_per_sec = Some(mib * 1024 * 1024); + } + "--checkpoint" => options.checkpoint = Some(PathBuf::from(value(arg)?)), + _ => return Err(format!("Unknown argument {arg}")), + } + } + Ok(options) +} + +pub async fn run(config: &Config, args: &[String]) -> i32 { + if args.iter().any(|a| a == "--help" || a == "-h") { + println!("{USAGE}"); + return 0; + } + let options = match parse(args) { + Ok(options) => options, + Err(message) => { + eprintln!("{message}\n\n{USAGE}"); + return 2; + } + }; + let Some(keys) = config.at_rest_key.as_deref() else { + eprintln!("SYNC_AT_REST_KEY is not set"); + return 1; + }; + let keyring = match Keyring::parse(keys) { + Ok(keyring) => Arc::new(keyring), + Err(message) => { + eprintln!("{message}"); + return 1; + } + }; + let store: Arc = match crate::raw_store(config) { + Ok(store) => Arc::new(store), + Err(message) => { + eprintln!("{message}"); + return 1; + } + }; + match migrate(store, keyring, &options).await { + Ok(report) if report.failed == 0 => 0, + Ok(_) => 1, + Err(message) => { + log::error(CONTEXT, &format!("Migration stopped err=\"{message}\"")); + 1 + } + } +} + +#[derive(Debug, Default, Clone, PartialEq)] +pub struct Report { + pub scanned: u64, + /// Already under the current key. + pub current: u64, + /// Plaintext objects sealed (or, in a dry run, to seal). + pub sealed: u64, + pub sealed_bytes: u64, + /// Objects under an older key re-sealed (or to re-seal). + pub rotated: u64, + pub rotated_bytes: u64, + /// Written or deleted by someone else meanwhile; left as they are. + pub changed: u64, + pub failed: u64, +} + +enum Outcome { + Current, + Sealed(u64), + Rotated(u64), + Changed, + Failed(String), +} + +/// Reads at most `per_sec` bytes a second, across every worker. +struct RateLimit { + per_sec: u64, + state: Mutex<(Instant, u64)>, +} + +impl RateLimit { + async fn take(&self, bytes: u64) { + let wait = { + let mut state = self.state.lock().unwrap(); + state.1 += bytes; + let due = Duration::from_secs_f64(state.1 as f64 / self.per_sec as f64); + due.saturating_sub(state.0.elapsed()) + }; + if !wait.is_zero() { + tokio::time::sleep(wait).await; + } + } +} + +fn throttled(body: ByteStream, limit: Option>) -> ByteStream { + match limit { + None => body, + Some(limit) => Box::pin(body.then(move |chunk| { + let limit = limit.clone(); + async move { + if let Ok(bytes) = &chunk { + limit.take(bytes.len() as u64).await; + } + chunk + } + })), + } +} + +struct Migration { + store: Arc, + keyring: Arc, + dry_run: bool, + conditional: bool, + limit: Option>, +} + +fn precondition_failed(error: &StoreError) -> bool { + matches!(error, StoreError::Service { status: 412, .. }) +} + +/// Whether the store refuses a PUT whose `If-Match` names another ETag. +pub async fn honors_if_match(store: &dyn Store) -> Result { + let probe = format!("{TMP_PREFIX}probe-{}", uuid::Uuid::new_v4().simple()); + store + .put_object(&probe, b"a".to_vec(), "text/plain", &[]) + .await?; + let outcome = store + .put_object_stream( + &probe, + crate::store::once(b"b".to_vec()), + Upload { + length: 1, + content_type: Some("text/plain"), + metadata: &[], + if_match: Some("\"00000000000000000000000000000000\""), + }, + ) + .await; + let _ = store.delete_object(&probe).await; + match outcome { + Ok(()) => Ok(false), + Err(e) if precondition_failed(&e) => Ok(true), + // Some stores (Backblaze B2) refuse the header instead of honoring it. + Err(StoreError::Service { .. }) => Ok(false), + Err(e) => Err(e), + } +} + +impl Migration { + async fn object(&self, key: &str) -> Outcome { + let mut attempt = 0; + loop { + attempt += 1; + match self.try_object(key).await { + Ok(outcome) => return outcome, + Err(StoreError::NotFound) => return Outcome::Changed, + Err(e) if precondition_failed(&e) => return Outcome::Changed, + Err(e @ StoreError::Corrupt(_)) => return Outcome::Failed(e.to_string()), + Err(e) if attempt >= ATTEMPTS => return Outcome::Failed(e.to_string()), + Err(_) => tokio::time::sleep(Duration::from_millis(500 << attempt)).await, + } + } + } + + async fn try_object(&self, key: &str) -> Result { + let head = self.store.head_object(key).await?; + let current = self.keyring.current().id(); + let old_key = match sealed_by(&head) { + Some(id) if id == current => return Ok(Outcome::Current), + Some(id) => Some(id.to_string()), + None => None, + }; + let stored = head.content_length.unwrap_or(0); + let plain = match &old_key { + Some(id) => at_rest::plain_len(id, stored).ok_or_else(|| { + StoreError::Corrupt(format!("sealed object has an invalid size {stored}")) + })?, + None => stored, + }; + let done = |plain| match old_key { + Some(_) => Outcome::Rotated(plain), + None => Outcome::Sealed(plain), + }; + if self.dry_run { + return Ok(done(plain)); + } + let etag = head + .etag + .clone() + .ok_or_else(|| StoreError::Transport("the store gave no ETag".to_string()))?; + let download = self.store.get_object(key).await?; + if download.head.etag.as_deref() != Some(etag.as_str()) { + return Ok(Outcome::Changed); + } + let mut body = throttled(download.body, self.limit.clone()); + if old_key.is_some() { + body = at_rest::open_stream(self.keyring.clone(), body); + } + let sealed = at_rest::seal_stream(&self.keyring, body); + let mut metadata: Vec<(String, String)> = head + .metadata + .iter() + .filter(|(name, _)| name != MARKER) + .cloned() + .collect(); + metadata.push((MARKER.to_string(), self.keyring.marker())); + let upload = Upload { + length: at_rest::sealed_len(current, plain), + content_type: head.content_type.as_deref(), + metadata: &metadata, + if_match: None, + }; + if self.conditional { + self + .store + .put_object_stream( + key, + sealed, + Upload { + if_match: Some(&etag), + ..upload + }, + ) + .await?; + return Ok(done(plain)); + } + let tmp = format!("{TMP_PREFIX}{}", uuid::Uuid::new_v4().simple()); + let placed = async { + self.store.put_object_stream(&tmp, sealed, upload).await?; + if self.store.head_object(key).await?.etag.as_deref() != Some(etag.as_str()) { + return Ok(false); + } + self.store.copy_object(&tmp, key).await?; + Ok::<_, StoreError>(true) + } + .await; + let _ = self.store.delete_object(&tmp).await; + Ok(if placed? { + done(plain) + } else { + Outcome::Changed + }) + } +} + +fn read_checkpoint(path: &PathBuf) -> Result, String> { + match std::fs::read_to_string(path) { + Ok(text) => Ok(Some(text.trim_end_matches('\n').to_string()).filter(|k| !k.is_empty())), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None), + Err(e) => Err(format!("Cannot read checkpoint {}: {e}", path.display())), + } +} + +fn write_checkpoint(path: &PathBuf, key: &str) -> Result<(), String> { + let tmp = path.with_extension("tmp"); + std::fs::write(&tmp, format!("{key}\n")) + .and_then(|()| std::fs::rename(&tmp, path)) + .map_err(|e| format!("Cannot write checkpoint {}: {e}", path.display())) +} + +fn summary(report: &Report, elapsed: Duration) -> String { + format!( + "scanned={} current={} sealed={} sealed_bytes={} rotated={} rotated_bytes={} changed={} failed={} elapsed_ms={}", + report.scanned, + report.current, + report.sealed, + report.sealed_bytes, + report.rotated, + report.rotated_bytes, + report.changed, + report.failed, + elapsed.as_millis() + ) +} + +/// Walks the bucket in key order. `Err` stops the walk: a listing or the +/// checkpoint failed. Objects that fail are counted, and keep the checkpoint +/// from moving past them. +pub async fn migrate( + store: Arc, + keyring: Arc, + options: &Options, +) -> Result { + let started = Instant::now(); + let conditional = if options.dry_run { + false + } else { + honors_if_match(store.as_ref()) + .await + .map_err(|e| format!("Cannot write to the bucket: {e}"))? + }; + let start_after = match &options.checkpoint { + Some(path) => read_checkpoint(path)?, + None => None, + }; + log::log( + CONTEXT, + &format!( + "Migration started key={} dry_run={} conditional_put={conditional} concurrency={} max_bytes_per_sec={} resume_after={}", + keyring.current().id(), + options.dry_run, + options.concurrency, + options.bytes_per_sec.unwrap_or(0), + start_after.as_deref().unwrap_or("-"), + ), + ); + let migration = Migration { + store: store.clone(), + keyring, + dry_run: options.dry_run, + conditional, + limit: options.bytes_per_sec.map(|per_sec| { + Arc::new(RateLimit { + per_sec, + state: Mutex::new((Instant::now(), 0)), + }) + }), + }; + let mut report = Report::default(); + let mut token: Option = None; + let mut last_progress = Instant::now(); + let mut checkpoint_held = false; + loop { + let page = store + .list_objects_v2(&ListInput { + prefix: options.prefix.clone(), + max_keys: Some("1000".to_string()), + continuation_token: token.clone(), + delimiter: None, + start_after: if token.is_none() { + start_after.clone() + } else { + None + }, + }) + .await + .map_err(|e| format!("Listing failed: {e}"))?; + let keys: Vec = page + .contents + .into_iter() + .map(|object| object.key) + .filter(|key| !key.starts_with(TMP_PREFIX)) + .collect(); + let outcomes: Vec<(String, Outcome)> = futures_util::stream::iter(keys.iter()) + .map(|key| async { (key.clone(), migration.object(key).await) }) + .buffer_unordered(options.concurrency) + .collect() + .await; + for (key, outcome) in outcomes { + report.scanned += 1; + match outcome { + Outcome::Current => report.current += 1, + Outcome::Sealed(bytes) => { + report.sealed += 1; + report.sealed_bytes += bytes; + } + Outcome::Rotated(bytes) => { + report.rotated += 1; + report.rotated_bytes += bytes; + } + Outcome::Changed => report.changed += 1, + Outcome::Failed(err) => { + report.failed += 1; + checkpoint_held = true; + log::warn( + CONTEXT, + &format!("Object not migrated key={key} err=\"{err}\""), + ); + } + } + } + if let (Some(path), Some(last), false, false) = ( + &options.checkpoint, + keys.last(), + checkpoint_held, + options.dry_run, + ) { + write_checkpoint(path, last)?; + } + token = page.next_continuation_token.filter(|_| page.is_truncated); + if token.is_none() { + break; + } + if last_progress.elapsed() >= PROGRESS_EVERY { + last_progress = Instant::now(); + log::log( + CONTEXT, + &format!( + "Migration progress {} last_key={}", + summary(&report, started.elapsed()), + keys.last().map_or("-", String::as_str) + ), + ); + } + } + let line = format!( + "{} {}", + if options.dry_run { + "Migration dry run done" + } else { + "Migration done" + }, + summary(&report, started.elapsed()) + ); + if report.failed > 0 { + log::warn(CONTEXT, &line); + } else { + log::log(CONTEXT, &line); + } + Ok(report) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::memory_store::MemoryStore; + use crate::sealed_store::SealedStore; + + const K1: &str = "k1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="; + const K2: &str = "k2:AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE="; + + fn options() -> Options { + Options { + concurrency: 3, + ..Options::default() + } + } + + fn meta(value: &str) -> Vec<(String, String)> { + vec![("updated-at".to_string(), value.to_string())] + } + + #[tokio::test] + async fn seals_plaintext_keeps_metadata_and_is_idempotent() { + let raw = Arc::new(MemoryStore::default()); + let big: Vec = (0..at_rest::FRAME * 2 + 9).map(|i| i as u8).collect(); + raw.insert( + "users/u/profiles/p/files/a", + big.clone(), + meta("5"), + Some("image/png".into()), + ); + raw.insert("users/u/proxies/x.json", b"{}".to_vec(), Vec::new(), None); + raw.insert("users/u/empty", Vec::new(), Vec::new(), None); + let keyring = Arc::new(Keyring::parse(K1).unwrap()); + + let dry = migrate( + raw.clone(), + keyring.clone(), + &Options { + dry_run: true, + ..options() + }, + ) + .await + .unwrap(); + assert_eq!( + (dry.scanned, dry.sealed, dry.sealed_bytes), + (3, 3, big.len() as u64 + 2) + ); + assert_eq!(raw.raw("users/u/proxies/x.json").unwrap().body, b"{}"); + + let report = migrate(raw.clone(), keyring.clone(), &options()) + .await + .unwrap(); + assert_eq!((report.sealed, report.failed, report.changed), (3, 0, 0)); + let sealed = SealedStore::new(raw.clone(), keyring.clone()); + let head = sealed + .head_object("users/u/profiles/p/files/a") + .await + .unwrap(); + assert_eq!(head.metadata, meta("5")); + assert_eq!(head.content_type.as_deref(), Some("image/png")); + assert_eq!(head.content_length, Some(big.len() as u64)); + let body = sealed + .get_object("users/u/profiles/p/files/a") + .await + .unwrap() + .body; + assert_eq!(crate::store::collect(body).await.unwrap(), big); + assert_eq!(sealed.get_object_text("users/u/empty").await.unwrap(), ""); + assert!(raw.keys().iter().all(|k| !k.starts_with(TMP_PREFIX))); + + let again = migrate(raw.clone(), keyring, &options()).await.unwrap(); + assert_eq!((again.current, again.sealed), (3, 0)); + } + + #[tokio::test] + async fn rotates_objects_under_an_older_key() { + let raw = Arc::new(MemoryStore::default()); + let old = Arc::new(Keyring::parse(K1).unwrap()); + SealedStore::new(raw.clone(), old) + .put_object("a", b"hello".to_vec(), "text/plain", &meta("1")) + .await + .unwrap(); + let rotated = Arc::new(Keyring::parse(&format!("{K2},{K1}")).unwrap()); + let report = migrate(raw.clone(), rotated.clone(), &options()) + .await + .unwrap(); + assert_eq!((report.rotated, report.rotated_bytes), (1, 5)); + let only_new = Arc::new(Keyring::parse(K2).unwrap()); + let store = SealedStore::new(raw.clone(), only_new); + assert_eq!(store.get_object_text("a").await.unwrap(), "hello"); + assert_eq!(store.head_object("a").await.unwrap().metadata, meta("1")); + } + + #[tokio::test] + async fn resumes_after_the_checkpoint() { + let raw = Arc::new(MemoryStore::default()); + for i in 0..2500 { + raw.insert(&format!("k{i:05}"), b"x".to_vec(), Vec::new(), None); + } + let keyring = Arc::new(Keyring::parse(K1).unwrap()); + let dir = std::env::temp_dir().join(format!("donut-migrate-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("checkpoint"); + std::fs::write(&path, "k01999\n").unwrap(); + let report = migrate( + raw.clone(), + keyring, + &Options { + checkpoint: Some(path.clone()), + ..options() + }, + ) + .await + .unwrap(); + assert_eq!(report.sealed, 500); + assert_eq!(std::fs::read_to_string(&path).unwrap(), "k02499\n"); + assert_eq!(raw.raw("k00000").unwrap().body, b"x"); + std::fs::remove_dir_all(dir).unwrap(); + } + + #[tokio::test] + async fn the_probe_sees_whether_if_match_is_honored() { + // MemoryStore refuses a PUT whose If-Match names another ETag. + let raw = Arc::new(MemoryStore::default()); + assert!(honors_if_match(raw.as_ref()).await.unwrap()); + assert!(raw.keys().is_empty(), "the probe cleans up"); + } + + #[test] + fn arguments_parse() { + let args: Vec = ["--dry-run", "--prefix", "users/", "--max-mib-per-sec", "8"] + .iter() + .map(|s| s.to_string()) + .collect(); + let options = parse(&args).unwrap(); + assert!(options.dry_run); + assert_eq!(options.prefix.as_deref(), Some("users/")); + assert_eq!(options.bytes_per_sec, Some(8 * 1024 * 1024)); + assert_eq!(options.concurrency, 4); + assert!(parse(&["--concurrency".to_string(), "0".to_string()]).is_err()); + assert!(parse(&["--bogus".to_string()]).is_err()); + } +} diff --git a/donut-sync/src/s3.rs b/donut-sync/src/s3.rs index dfd2519d..4c7eb331 100644 --- a/donut-sync/src/s3.rs +++ b/donut-sync/src/s3.rs @@ -10,7 +10,9 @@ use reqwest::Method; use crate::jsval::js_number; use crate::sigv4::{self, Credentials}; -use crate::store::{HeadOutput, ListInput, ListObject, ListOutput, Store, StoreError}; +use crate::store::{ + ByteStream, Download, HeadOutput, ListInput, ListObject, ListOutput, Store, StoreError, Upload, +}; use crate::time; const MAX_ATTEMPTS: u32 = 3; @@ -377,16 +379,16 @@ impl S3Store { } } - async fn attempt( + /// A signed request, without its body. + fn request( &self, method: &Method, path: &str, query: &[(String, String)], headers: &[(String, String)], - body: &[u8], - ) -> Result { + payload_hash: String, + ) -> reqwest::RequestBuilder { let amz_date = time::compact_stamp(time::now_ms()); - let payload_hash = sigv4::sha256_hex(body); let mut signed = vec![ ("host".to_string(), self.location.request_host()), ("x-amz-content-sha256".to_string(), payload_hash.clone()), @@ -414,6 +416,18 @@ impl S3Store { for (name, value) in signed.iter().filter(|(name, _)| name != "host") { request = request.header(name.as_str(), value.as_str()); } + request + } + + async fn attempt( + &self, + method: &Method, + path: &str, + query: &[(String, String)], + headers: &[(String, String)], + body: &[u8], + ) -> Result { + let mut request = self.request(method, path, query, headers, sigv4::sha256_hex(body)); if !body.is_empty() || *method == Method::PUT || *method == Method::POST { request = request.body(body.to_vec()); } @@ -487,6 +501,44 @@ impl S3Store { } } +/// What a HEAD or GET says about the object. +fn head_of(headers: &reqwest::header::HeaderMap) -> HeadOutput { + let header = |name: &str| headers.get(name).map(|v| latin1(v.as_bytes())); + let mut metadata: Vec<(String, String)> = headers + .iter() + .filter_map(|(name, value)| { + name + .as_str() + .strip_prefix("x-amz-meta-") + .map(|n| (n.to_string(), latin1(value.as_bytes()))) + }) + .collect(); + metadata.sort_by(|a, b| a.0.cmp(&b.0)); + HeadOutput { + last_modified_ms: header("last-modified").and_then(|v| time::parse_http_date_ms(&v)), + content_length: header("content-length").and_then(|v| v.trim().parse().ok()), + metadata, + etag: header("etag"), + content_type: header("content-type"), + } +} + +/// A refusal, read from a streamed response. +async fn refusal(response: reqwest::Response) -> StoreError { + let status = response.status().as_u16(); + let headers = response.headers().clone(); + let body = response + .bytes() + .await + .map(|b| b.to_vec()) + .unwrap_or_default(); + error_of(&Reply { + status, + headers, + body, + }) +} + fn transport_message(error: &reqwest::Error) -> String { let mut message = error.to_string(); let mut source = std::error::Error::source(error); @@ -521,24 +573,7 @@ impl Store for S3Store { let reply = self .expect_success(Method::HEAD, &path, &[], &[], &[]) .await?; - let header = |name: &str| reply.headers.get(name).map(|v| latin1(v.as_bytes())); - let mut metadata: Vec<(String, String)> = reply - .headers - .iter() - .filter_map(|(name, value)| { - name - .as_str() - .strip_prefix("x-amz-meta-") - .map(|n| (n.to_string(), latin1(value.as_bytes()))) - }) - .collect(); - metadata.sort_by(|a, b| a.0.cmp(&b.0)); - Ok(HeadOutput { - last_modified_ms: header("last-modified").and_then(|v| time::parse_http_date_ms(&v)), - content_length: header("content-length").and_then(|v| v.trim().parse().ok()), - metadata, - etag: header("etag"), - }) + Ok(head_of(&reply.headers)) } async fn get_object_text(&self, key: &str) -> Result { @@ -554,9 +589,13 @@ impl Store for S3Store { key: &str, body: Vec, content_type: &str, + metadata: &[(String, String)], ) -> Result<(), StoreError> { let path = self.location.object_path(key); - let headers = [("content-type".to_string(), content_type.to_string())]; + let mut headers = vec![("content-type".to_string(), content_type.to_string())]; + for (name, value) in metadata { + headers.push((format!("x-amz-meta-{name}"), value.clone())); + } self .expect_success(Method::PUT, &path, &[], &headers, &body) .await @@ -640,12 +679,82 @@ impl Store for S3Store { if let Some(prefix) = &input.prefix { query.push(("prefix".to_string(), prefix.clone())); } + if let Some(start_after) = &input.start_after { + query.push(("start-after".to_string(), start_after.clone())); + } let path = self.location.bucket_path(); let reply = self .expect_success(Method::GET, &path, &query, &[], &[]) .await?; parse_list(&reply.body) } + + async fn get_object(&self, key: &str) -> Result { + let path = self.location.object_path(key); + let mut attempt = 0; + let response = loop { + attempt += 1; + let sent = self + .request(&Method::GET, &path, &[], &[], sigv4::sha256_hex(&[])) + .send() + .await; + let retry = match &sent { + Err(_) => true, + Ok(response) => matches!(response.status().as_u16(), 429 | 500 | 502 | 503 | 504), + }; + if !retry || attempt >= MAX_ATTEMPTS { + break sent.map_err(|e| StoreError::Transport(transport_message(&e)))?; + } + tokio::time::sleep(RETRY_BASE.saturating_mul(1 << attempt)).await; + }; + if !response.status().is_success() { + return Err(refusal(response).await); + } + let head = head_of(response.headers()); + let body = futures_util::StreamExt::map(response.bytes_stream(), |chunk| { + chunk.map_err(|e| StoreError::Transport(transport_message(&e))) + }); + Ok(Download { + head, + body: Box::pin(body), + }) + } + + /// One attempt: a stream cannot be sent twice. + async fn put_object_stream( + &self, + key: &str, + body: ByteStream, + upload: Upload<'_>, + ) -> Result<(), StoreError> { + let path = self.location.object_path(key); + let mut headers = vec![("content-length".to_string(), upload.length.to_string())]; + if let Some(content_type) = upload.content_type { + headers.push(("content-type".to_string(), content_type.to_string())); + } + if let Some(etag) = upload.if_match { + headers.push(("if-match".to_string(), etag.to_string())); + } + for (name, value) in upload.metadata { + headers.push((format!("x-amz-meta-{name}"), value.clone())); + } + let response = self + .request( + &Method::PUT, + &path, + &[], + &headers, + sigv4::UNSIGNED_PAYLOAD.to_string(), + ) + .body(reqwest::Body::wrap_stream(body)) + .send() + .await + .map_err(|e| StoreError::Transport(transport_message(&e)))?; + if !response.status().is_success() { + return Err(refusal(response).await); + } + Ok(()) + } } #[cfg(test)] diff --git a/donut-sync/src/sealed_store.rs b/donut-sync/src/sealed_store.rs new file mode 100644 index 00000000..722169ca --- /dev/null +++ b/donut-sync/src/sealed_store.rs @@ -0,0 +1,133 @@ +//! The store as the routes see it when `SYNC_AT_REST_KEY` is set: writes are +//! sealed, reads opened, and plaintext objects from before still read as they +//! are. HEAD answers with the plaintext size and without the marker. + +use std::sync::Arc; + +use async_trait::async_trait; + +use crate::at_rest::{self, Keyring, MARKER}; +use crate::store::{ + ByteStream, Download, HeadOutput, ListInput, ListOutput, Store, StoreError, Upload, +}; + +pub struct SealedStore { + inner: Arc, + keyring: Arc, +} + +/// The key id a stored object is sealed under; `None` for plaintext. +pub fn sealed_by(head: &HeadOutput) -> Option<&str> { + head.metadata_value(MARKER).and_then(at_rest::marker_key_id) +} + +/// `head` as the plaintext object would have it. +fn reveal(mut head: HeadOutput) -> Result { + if let Some(id) = sealed_by(&head) { + let id = id.to_string(); + head.content_length = match head.content_length { + Some(sealed) => Some(at_rest::plain_len(&id, sealed).ok_or_else(|| { + StoreError::Corrupt(format!("sealed object has an invalid size {sealed}")) + })?), + None => None, + }; + head.metadata.retain(|(name, _)| name != MARKER); + } + Ok(head) +} + +impl SealedStore { + pub fn new(inner: Arc, keyring: Arc) -> Self { + Self { inner, keyring } + } + + fn with_marker(&self, metadata: &[(String, String)]) -> Vec<(String, String)> { + let mut out = metadata.to_vec(); + out.push((MARKER.to_string(), self.keyring.marker())); + out + } +} + +#[async_trait] +impl Store for SealedStore { + async fn head_bucket(&self) -> Result<(), StoreError> { + self.inner.head_bucket().await + } + + async fn create_bucket(&self) -> Result<(), StoreError> { + self.inner.create_bucket().await + } + + async fn head_object(&self, key: &str) -> Result { + reveal(self.inner.head_object(key).await?) + } + + async fn get_object_text(&self, key: &str) -> Result { + let body = crate::store::collect(self.get_object(key).await?.body).await?; + Ok(String::from_utf8_lossy(&body).into_owned()) + } + + async fn put_object( + &self, + key: &str, + body: Vec, + content_type: &str, + metadata: &[(String, String)], + ) -> Result<(), StoreError> { + let sealed = at_rest::seal(&self.keyring, &body); + self + .inner + .put_object(key, sealed, content_type, &self.with_marker(metadata)) + .await + } + + async fn delete_object(&self, key: &str) -> Result<(), StoreError> { + self.inner.delete_object(key).await + } + + async fn delete_objects(&self, keys: &[String]) -> Result<(), StoreError> { + self.inner.delete_objects(keys).await + } + + /// Sealed objects copy as they are: the frames do not name their key. + async fn copy_object(&self, from: &str, to: &str) -> Result<(), StoreError> { + self.inner.copy_object(from, to).await + } + + async fn list_objects_v2(&self, input: &ListInput) -> Result { + self.inner.list_objects_v2(input).await + } + + async fn get_object(&self, key: &str) -> Result { + let download = self.inner.get_object(key).await?; + let sealed = sealed_by(&download.head).is_some(); + let head = reveal(download.head)?; + let body = if sealed { + at_rest::open_stream(self.keyring.clone(), download.body) + } else { + download.body + }; + Ok(Download { head, body }) + } + + async fn put_object_stream( + &self, + key: &str, + body: ByteStream, + upload: Upload<'_>, + ) -> Result<(), StoreError> { + let metadata = self.with_marker(upload.metadata); + self + .inner + .put_object_stream( + key, + at_rest::seal_stream(&self.keyring, body), + Upload { + length: at_rest::sealed_len(self.keyring.current().id(), upload.length), + metadata: &metadata, + ..upload + }, + ) + .await + } +} diff --git a/donut-sync/src/store.rs b/donut-sync/src/store.rs index 04c7973d..f9acb141 100644 --- a/donut-sync/src/store.rs +++ b/donut-sync/src/store.rs @@ -1,6 +1,12 @@ //! The object storage operations the server uses. +use std::pin::Pin; + use async_trait::async_trait; +use bytes::Bytes; +use futures_util::Stream; + +pub type ByteStream = Pin> + Send>>; #[derive(Debug, Clone, Default)] pub struct HeadOutput { @@ -9,6 +15,7 @@ pub struct HeadOutput { /// User metadata: lowercase names without `x-amz-meta-`, in response order. pub metadata: Vec<(String, String)>, pub etag: Option, + pub content_type: Option, } impl HeadOutput { @@ -27,6 +34,7 @@ pub struct ListInput { pub max_keys: Option, pub continuation_token: Option, pub delimiter: Option, + pub start_after: Option, } #[derive(Debug, Clone)] @@ -56,6 +64,8 @@ pub enum StoreError { }, /// The store could not be reached, or answered with something unreadable. Transport(String), + /// A sealed object that does not open: forged, cut off, or under an unknown key. + Corrupt(String), } impl StoreError { @@ -83,6 +93,7 @@ impl std::fmt::Display for StoreError { } } StoreError::Transport(message) => write!(f, "{message}"), + StoreError::Corrupt(message) => write!(f, "{message}"), } } } @@ -100,6 +111,7 @@ pub trait Store: Send + Sync { key: &str, body: Vec, content_type: &str, + metadata: &[(String, String)], ) -> Result<(), StoreError>; async fn delete_object(&self, key: &str) -> Result<(), StoreError>; /// Deletes up to 1000 keys in one request; per-key failures are ignored. @@ -107,4 +119,44 @@ pub trait Store: Send + Sync { /// Server-side copy, metadata included. async fn copy_object(&self, from: &str, to: &str) -> Result<(), StoreError>; async fn list_objects_v2(&self, input: &ListInput) -> Result; + /// The object's body as it arrives, with what a HEAD would say. + async fn get_object(&self, key: &str) -> Result; + /// Writes `upload.length` bytes from `body`. With `if_match`, the store + /// refuses (412) when the object's ETag differs, where it supports that. + async fn put_object_stream( + &self, + key: &str, + body: ByteStream, + upload: Upload<'_>, + ) -> Result<(), StoreError>; +} + +pub struct Download { + pub head: HeadOutput, + pub body: ByteStream, +} + +#[derive(Clone, Copy)] +pub struct Upload<'a> { + pub length: u64, + pub content_type: Option<&'a str>, + /// Names without `x-amz-meta-`. + pub metadata: &'a [(String, String)], + pub if_match: Option<&'a str>, +} + +/// All of a stream, in memory. +pub async fn collect(mut body: ByteStream) -> Result, StoreError> { + use futures_util::StreamExt; + let mut out = Vec::new(); + while let Some(chunk) = body.next().await { + out.extend_from_slice(&chunk?); + } + Ok(out) +} + +pub fn once(bytes: Vec) -> ByteStream { + Box::pin(futures_util::stream::once( + async move { Ok(Bytes::from(bytes)) }, + )) } diff --git a/donut-sync/src/subscribe.rs b/donut-sync/src/subscribe.rs index 618a8b57..9c2f7255 100644 --- a/donut-sync/src/subscribe.rs +++ b/donut-sync/src/subscribe.rs @@ -116,6 +116,7 @@ async fn poll_once( max_keys: Some("1000".to_string()), continuation_token: token.clone(), delimiter: None, + start_after: None, }) .await { diff --git a/donut-sync/src/sync.rs b/donut-sync/src/sync.rs index 1c3861c4..9d390886 100644 --- a/donut-sync/src/sync.rs +++ b/donut-sync/src/sync.rs @@ -17,6 +17,7 @@ use crate::log; use crate::s3::Presigner; use crate::store::{HeadOutput, ListInput, Store, StoreError}; use crate::time; +use crate::transfer::TransferSigner; const CONTEXT: &str = "SyncService"; @@ -78,6 +79,12 @@ fn clamp_expires_in(requested: Field) -> f64 { value.min(MAX_PRESIGN_EXPIRES_IN) } +/// When a URL issued at `now` for `expires_in` seconds stops working, as +/// `expiresAt` reports it. +fn expires_ms(now: i64, expires_in: f64) -> i64 { + (now as f64 + expires_in * 1000.0).trunc() as i64 +} + /// `new Date(ms).toISOString()` for a fractional millisecond time. fn iso_at(ms: f64) -> String { time::iso_from_ms(ms.trunc() as i64) @@ -147,6 +154,7 @@ fn group_denied() -> ApiError { pub struct SyncService { store: Arc, presigner: Presigner, + transfer: Option, public_endpoint: String, public_endpoint_configured: bool, cloud: bool, @@ -156,6 +164,8 @@ pub struct SyncService { pub struct SyncOptions { pub presigner: Presigner, + /// Set when the store is sealed at rest: transfers go through this server. + pub transfer: Option, pub public_endpoint: String, pub public_endpoint_configured: bool, /// `SYNC_JWT_PUBLIC_KEY` is set: Donut's own deployment. @@ -168,6 +178,7 @@ impl SyncService { Self { store, presigner: options.presigner, + transfer: options.transfer, public_endpoint: options.public_endpoint, public_endpoint_configured: options.public_endpoint_configured, cloud: options.cloud, @@ -180,6 +191,31 @@ impl SyncService { &self.store } + pub fn transfer(&self) -> Option<&TransferSigner> { + self.transfer.as_ref() + } + + fn put_url( + &self, + origin: &str, + key: &str, + expires_in: f64, + metadata: &[(String, String)], + now: i64, + ) -> String { + match &self.transfer { + Some(signer) => signer.url(origin, "PUT", key, expires_ms(now, expires_in), metadata), + None => self.presigner.put_url(key, expires_in, metadata, now), + } + } + + fn get_url(&self, origin: &str, key: &str, expires_in: f64, now: i64) -> String { + match &self.transfer { + Some(signer) => signer.url(origin, "GET", key, expires_ms(now, expires_in), &[]), + None => self.presigner.get_url(key, expires_in, now), + } + } + // --- startup and probes ---------------------------------------------------- pub async fn ensure_bucket_exists(&self) -> Result<(), String> { @@ -202,14 +238,18 @@ impl SyncService { /// The address clients are sent to for transfers, for `/readyz`. Withheld /// in cloud mode: `/readyz` is public, and a managed deployment should not /// publish its storage host. + /// Withheld too when transfers come here: devices never reach storage. pub fn diagnostic_storage_endpoint(&self) -> Option<&str> { - (!self.cloud).then_some(self.public_endpoint.as_str()) + (!self.cloud && self.transfer.is_none()).then_some(self.public_endpoint.as_str()) } /// A single-label host (`minio`, `s3`) resolves only inside a container /// network, so every presigned URL built from it fails for the desktop /// client while the server's own S3 calls succeed. Say it once at boot. pub fn warn_if_presign_endpoint_is_server_only(&self) { + if self.transfer.is_some() { + return; + } let Ok(url) = url::Url::parse(&self.public_endpoint) else { return; }; @@ -427,7 +467,7 @@ impl SyncService { }) .to_string(); if let Err(e) = store - .put_object(&key, body.into_bytes(), "application/json") + .put_object(&key, body.into_bytes(), "application/json", &[]) .await { log::warn(CONTEXT, &format!("Manifest bump failed for {key}: {e}")); @@ -448,7 +488,12 @@ impl SyncService { } } - pub async fn presign_upload(&self, body: &Body, ctx: &UserContext) -> Result { + pub async fn presign_upload( + &self, + body: &Body, + ctx: &UserContext, + origin: &str, + ) -> Result { let raw_key = body.field("key")?; let key = self.scope_key(ctx, raw_key); self.validate_write_access(ctx, key.as_deref())?; @@ -463,7 +508,8 @@ impl SyncService { let expires_in = clamp_expires_in(body.field("expiresIn")?); let now = time::now_ms(); let expires_at = iso_at(now as f64 + expires_in * 1000.0); - let url = self.presigner.put_url( + let url = self.put_url( + origin, require_key(&key)?, expires_in, metadata.as_deref().unwrap_or(&[]), @@ -484,14 +530,19 @@ impl SyncService { Ok(Value::Object(out)) } - pub async fn presign_download(&self, body: &Body, ctx: &UserContext) -> Result { + pub async fn presign_download( + &self, + body: &Body, + ctx: &UserContext, + origin: &str, + ) -> Result { let key = self.scope_key(ctx, body.field("key")?); self.validate_key_access(ctx, key.as_deref())?; self.assert_reachable(ctx, key.as_deref()).await?; let expires_in = clamp_expires_in(body.field("expiresIn")?); let now = time::now_ms(); let expires_at = iso_at(now as f64 + expires_in * 1000.0); - let url = self.presigner.get_url(require_key(&key)?, expires_in, now); + let url = self.get_url(origin, require_key(&key)?, expires_in, now); Ok(json!({ "url": url, "expiresAt": expires_at })) } @@ -538,6 +589,7 @@ impl SyncService { require_key(&tombstone_key)?, Value::Object(record).to_string().into_bytes(), "application/json", + &[], ) .await .map_err(store_error)?; @@ -571,6 +623,7 @@ impl SyncService { }), continuation_token: token.to_key(), delimiter: None, + start_after: None, }) .await .map_err(store_error)?; @@ -615,6 +668,7 @@ impl SyncService { &self, body: &Body, ctx: &UserContext, + origin: &str, ) -> Result { let items = body .field("items")? @@ -655,9 +709,7 @@ impl SyncService { .await?; let mut out_items = Vec::with_capacity(keys.len()); for (raw, key) in raw_keys.iter().zip(&keys) { - let url = self - .presigner - .put_url(require_key(key)?, expires_in, &[], now); + let url = self.put_url(origin, require_key(key)?, expires_in, &[], now); let mut item = Map::new(); if let Field::Value(value) = raw { item.insert("key".to_string(), (*value).clone()); @@ -695,6 +747,7 @@ impl SyncService { &self, body: &Body, ctx: &UserContext, + origin: &str, ) -> Result { let raw_keys = body .field("keys")? @@ -724,7 +777,7 @@ impl SyncService { .await?; let mut items = Vec::with_capacity(keys.len()); for (raw, key) in raw_keys.iter().zip(&keys) { - let url = self.presigner.get_url(require_key(key)?, expires_in, now); + let url = self.get_url(origin, require_key(key)?, expires_in, now); items.push(json!({ "key": raw, "url": url, "expiresAt": expires_at })); } Ok(json!({ "items": items })) @@ -762,6 +815,7 @@ impl SyncService { max_keys: Some("1000".to_string()), continuation_token: token.clone(), delimiter: None, + start_after: None, }) .await .map_err(store_error)?; @@ -794,6 +848,7 @@ impl SyncService { require_key(&tombstone_key)?, Value::Object(record).to_string().into_bytes(), "application/json", + &[], ) .await .map_err(store_error)?; @@ -837,6 +892,7 @@ impl SyncService { max_keys: None, continuation_token: token.clone(), delimiter: None, + start_after: None, }) .await .map_err(store_error)?; @@ -977,6 +1033,7 @@ impl SyncService { &format!("{target}snapshot.json"), snapshot.to_string().into_bytes(), "application/json", + &[], ) .await .map_err(store_error)?; @@ -1108,6 +1165,7 @@ impl SyncService { max_keys: Some("1000".to_string()), continuation_token: token.clone(), delimiter: Some("/".to_string()), + start_after: None, }) .await .map_err(store_error)?; @@ -1125,6 +1183,7 @@ impl SyncService { max_keys: Some("1".to_string()), continuation_token: None, delimiter: None, + start_after: None, }) .await .map_err(store_error)?; @@ -1160,6 +1219,7 @@ impl SyncService { max_keys: Some("1000".to_string()), continuation_token: token.clone(), delimiter: None, + start_after: None, }) .await .map_err(store_error)?; @@ -1187,6 +1247,7 @@ impl SyncService { &format!("{user_prefix}tombstones/profiles/{id}"), record.to_string().into_bytes(), "application/json", + &[], ) .await .map_err(store_error)?; @@ -1290,6 +1351,7 @@ async fn stored_profile_ids(store: &dyn Store, prefix: &str) -> Result Result + Send + Sync>; -type ListHook = Box ListOutput + Send + Sync>; - -#[derive(Clone)] -struct Object { - body: Vec, - metadata: Vec<(String, String)>, -} - -/// A bucket in memory. Hooks stand in for HEAD and LIST where a test needs -/// answers the bucket itself would not give. -#[derive(Default)] -struct MemoryStore { - objects: Mutex>, - calls: AtomicUsize, - head_hook: Option, - list_hook: Option, -} - -impl MemoryStore { - fn call(&self) { - self.calls.fetch_add(1, Ordering::SeqCst); - } - - fn set(&self, key: &str, body: &str) { - self.objects.lock().unwrap().insert( - key.to_string(), - Object { - body: body.as_bytes().to_vec(), - metadata: Vec::new(), - }, - ); - } - - fn get(&self, key: &str) -> Option { - self - .objects - .lock() - .unwrap() - .get(key) - .map(|o| String::from_utf8_lossy(&o.body).into_owned()) - } - - fn keys(&self) -> Vec { - self.objects.lock().unwrap().keys().cloned().collect() - } -} - -#[async_trait] -impl Store for MemoryStore { - async fn head_bucket(&self) -> Result<(), StoreError> { - self.call(); - Ok(()) - } - - async fn create_bucket(&self) -> Result<(), StoreError> { - self.call(); - Ok(()) - } - - async fn head_object(&self, key: &str) -> Result { - self.call(); - if let Some(hook) = &self.head_hook { - return hook(key); - } - let objects = self.objects.lock().unwrap(); - let object = objects.get(key).ok_or(StoreError::NotFound)?; - Ok(HeadOutput { - last_modified_ms: Some(0), - content_length: Some(object.body.len() as u64), - metadata: object.metadata.clone(), - etag: Some(format!("\"{}\"", object.body.len())), - }) - } - - async fn get_object_text(&self, key: &str) -> Result { - self.call(); - self.get(key).ok_or(StoreError::NotFound) - } - - async fn put_object( - &self, - key: &str, - body: Vec, - _content_type: &str, - ) -> Result<(), StoreError> { - self.call(); - self.objects.lock().unwrap().insert( - key.to_string(), - Object { - body, - metadata: Vec::new(), - }, - ); - Ok(()) - } - - async fn delete_object(&self, key: &str) -> Result<(), StoreError> { - self.call(); - self.objects.lock().unwrap().remove(key); - Ok(()) - } - - async fn delete_objects(&self, keys: &[String]) -> Result<(), StoreError> { - self.call(); - let mut objects = self.objects.lock().unwrap(); - for key in keys { - objects.remove(key); - } - Ok(()) - } - - async fn copy_object(&self, from: &str, to: &str) -> Result<(), StoreError> { - self.call(); - let mut objects = self.objects.lock().unwrap(); - let object = objects.get(from).cloned().ok_or(StoreError::NotFound)?; - objects.insert(to.to_string(), object); - Ok(()) - } - - async fn list_objects_v2(&self, input: &ListInput) -> Result { - self.call(); - if let Some(hook) = &self.list_hook { - return Ok(hook(input)); - } - let prefix = input.prefix.clone().unwrap_or_default(); - let objects = self.objects.lock().unwrap(); - let mut out = ListOutput::default(); - for (key, object) in objects.range(prefix.clone()..) { - if !key.starts_with(&prefix) { - break; - } - if let Some(delimiter) = &input.delimiter - && let Some(at) = key[prefix.len()..].find(delimiter.as_str()) - { - let common = key[..prefix.len() + at + delimiter.len()].to_string(); - if !out.common_prefixes.contains(&common) { - out.common_prefixes.push(common); - } - continue; - } - out.contents.push(ListObject { - key: key.clone(), - last_modified_ms: Some(0), - size: Some(object.body.len() as u64), - }); - } - Ok(out) - } -} - fn service_with(store: Arc) -> SyncService { let location = crate::s3::Location::parse("http://s3.invalid:9000", "test-bucket", true).unwrap(); let presigner = Presigner::new( @@ -174,6 +24,7 @@ fn service_with(store: Arc) -> SyncService { store, SyncOptions { presigner, + transfer: None, public_endpoint: "http://s3.invalid:9000".to_string(), public_endpoint_configured: false, cloud: true, @@ -216,7 +67,7 @@ async fn a_launcher_syncs_what_a_session_changed() { "teams/t1/profiles/p1/manifest.json", ] { let out = service - .presign_upload(&body(json!({ "key": key })), &launcher) + .presign_upload(&body(json!({ "key": key })), &launcher, ORIGIN) .await .unwrap(); assert!(out["url"].is_string()); @@ -225,6 +76,7 @@ async fn a_launcher_syncs_what_a_session_changed() { .presign_upload_batch( &body(json!({ "items": [{ "key": "profiles/p1/files/Default/Preferences" }] })), &launcher, + ORIGIN, ) .await .unwrap(); @@ -257,7 +109,7 @@ async fn every_other_launcher_write_is_refused_before_it_reaches_s3() { assert_eq!( refusal_code( service - .presign_upload(&body(json!({ "key": key })), &launcher) + .presign_upload(&body(json!({ "key": key })), &launcher, ORIGIN) .await ), "TEAM_ROLE_READ_ONLY", @@ -273,6 +125,7 @@ async fn every_other_launcher_write_is_refused_before_it_reaches_s3() { { "key": "profiles/p1/metadata.json" }, ] })), &launcher, + ORIGIN, ) .await ), @@ -311,6 +164,7 @@ async fn a_member_changes_profile_settings_and_deletes_profiles() { .presign_upload( &body(json!({ "key": "profiles/p1/metadata.json" })), &member, + ORIGIN, ) .await .unwrap(); @@ -402,7 +256,7 @@ async fn a_limited_member_sees_only_the_profiles_and_groups_it_may_reach() { ); } let out = service - .presign_download(&body(json!({ "key": "proxies/x.json" })), &ctx) + .presign_download(&body(json!({ "key": "proxies/x.json" })), &ctx, ORIGIN) .await .unwrap(); assert!(out["url"].is_string()); @@ -436,6 +290,7 @@ async fn a_limited_member_changes_profiles_only_inside_its_groups() { .presign_upload( &body(json!({ "key": "profiles/new/metadata.json", "metadata": { "group-id": "g1" } })), &ctx, + ORIGIN, ) .await .unwrap(); @@ -443,6 +298,7 @@ async fn a_limited_member_changes_profiles_only_inside_its_groups() { .presign_upload( &body(json!({ "key": "profiles/p1/files/Default/Cookies" })), &ctx, + ORIGIN, ) .await .unwrap(); @@ -455,7 +311,11 @@ async fn a_limited_member_changes_profiles_only_inside_its_groups() { json!({ "key": "groups/g2.json" }), ] { assert_eq!( - refusal_code(service.presign_upload(&body(request.clone()), &ctx).await), + refusal_code( + service + .presign_upload(&body(request.clone()), &ctx, ORIGIN) + .await + ), "TEAM_GROUP_ACCESS_DENIED", "{request}" ); @@ -667,12 +527,16 @@ async fn a_full_plan_keeps_syncing_its_profiles_and_settings_but_adds_none() { "groups/g.json", ] { service - .presign_upload(&body(json!({ "key": key })), &ctx) + .presign_upload(&body(json!({ "key": key })), &ctx, ORIGIN) .await .unwrap(); } let error = service - .presign_upload(&body(json!({ "key": "profiles/p3/metadata.json" })), &ctx) + .presign_upload( + &body(json!({ "key": "profiles/p3/metadata.json" })), + &ctx, + ORIGIN, + ) .await .unwrap_err(); assert_eq!(error.status(), 403); @@ -686,6 +550,7 @@ async fn a_full_plan_keeps_syncing_its_profiles_and_settings_but_adds_none() { json!({ "items": [{ "key": "profiles/p1/files/a" }, { "key": "profiles/p3/files/a" }] }), ), &ctx, + ORIGIN, ) .await .unwrap_err(); @@ -702,6 +567,7 @@ async fn a_batch_counts_its_new_profiles_together() { json!({ "items": [{ "key": "profiles/p2/files/a" }, { "key": "profiles/p2/files/b" }] }), ), &ctx, + ORIGIN, ) .await .unwrap(); @@ -711,6 +577,7 @@ async fn a_batch_counts_its_new_profiles_together() { json!({ "items": [{ "key": "profiles/p2/files/a" }, { "key": "profiles/p3/files/a" }] }), ), &ctx, + ORIGIN, ) .await .unwrap_err(); @@ -723,20 +590,21 @@ async fn the_profile_limit_applies_in_the_cloud_only() { let ctx = cloud_with_limit(); assert_eq!( service - .presign_upload(&body(json!({ "key": "profiles/p1/files/a" })), &ctx) + .presign_upload(&body(json!({ "key": "profiles/p1/files/a" })), &ctx, ORIGIN) .await .unwrap_err() .status(), 403 ); service - .presign_upload(&body(json!({ "key": "proxies/x.json" })), &ctx) + .presign_upload(&body(json!({ "key": "proxies/x.json" })), &ctx, ORIGIN) .await .unwrap(); service .presign_upload( &body(json!({ "key": "profiles/p9/files/a" })), &UserContext::self_hosted(), + ORIGIN, ) .await .unwrap(); @@ -747,6 +615,7 @@ async fn the_profile_limit_applies_in_the_cloud_only() { mode: Mode::Account, ..ctx }, + ORIGIN, ) .await .unwrap(); @@ -807,6 +676,7 @@ async fn presigned_uploads_echo_only_the_metadata_sync_uses() { "expiresIn": 999999, })), &UserContext::self_hosted(), + ORIGIN, ) .await .unwrap(); diff --git a/donut-sync/src/time.rs b/donut-sync/src/time.rs index 1ba7452d..1eb48687 100644 --- a/donut-sync/src/time.rs +++ b/donut-sync/src/time.rs @@ -156,6 +156,10 @@ pub fn parse_http_date_ms(s: &str) -> Option { .map(|d| d.as_millis() as i64) } +pub fn http_date(ms: i64) -> String { + httpdate::fmt_http_date(UNIX_EPOCH + std::time::Duration::from_millis(ms.max(0) as u64)) +} + #[cfg(test)] mod tests { use super::*; diff --git a/donut-sync/src/transfer.rs b/donut-sync/src/transfer.rs new file mode 100644 index 00000000..a7561d6a --- /dev/null +++ b/donut-sync/src/transfer.rs @@ -0,0 +1,395 @@ +//! Transfers through this server, for a store sealed at rest. Devices cannot +//! go to storage directly, so the presign routes hand out URLs to `PATH` here, +//! signed and expiring like S3's, and the body streams through the sealed +//! store. A device sends and reads exactly what it did with S3: the same +//! method, `Content-Length`, `Content-Type` and the signed `x-amz-meta-*`. + +use std::sync::Arc; + +use bytes::Bytes; +use futures_util::StreamExt; +use hmac::{Hmac, KeyInit, Mac}; +use http_body_util::{BodyExt, StreamBody}; +use hyper::body::{Frame, Incoming}; +use hyper::http::request::Parts; +use hyper::{Method, Response, header}; +use sha2::Sha256; + +use crate::at_rest::Keyring; +use crate::http::{BoxError, ResBody, drain, full}; +use crate::store::{Store, StoreError, Upload}; +use crate::time; + +pub const PATH: &str = "/v1/transfer"; +/// Transfers streamed at a time; more wait their turn. Each holds a few +/// frames in memory. +const MAX_TRANSFERS: usize = 64; +const META_PARAM: &str = "meta-"; + +pub struct TransferSigner { + /// The current key's first; any of them verifies. + keys: Vec<[u8; 32]>, + slots: Arc, +} + +/// What a valid URL lets its holder do. +#[derive(Debug, PartialEq)] +pub struct Grant { + pub key: String, + pub metadata: Vec<(String, String)>, +} + +#[derive(Debug, PartialEq)] +pub enum Refusal { + Expired, + BadSignature, +} + +fn field(mac: &mut Hmac, value: &str) { + mac.update(&(value.len() as u64).to_be_bytes()); + mac.update(value.as_bytes()); +} + +impl TransferSigner { + pub fn new(keyring: &Keyring) -> Self { + Self { + keys: keyring.url_keys(), + slots: Arc::new(tokio::sync::Semaphore::new(MAX_TRANSFERS)), + } + } + + fn mac( + key: &[u8; 32], + method: &str, + object_key: &str, + expires_ms: &str, + metadata: &[(String, String)], + ) -> Hmac { + let mut mac = Hmac::::new_from_slice(key).expect("HMAC takes any key length"); + for value in ["v1", method, object_key, expires_ms] { + field(&mut mac, value); + } + let mut sorted = metadata.to_vec(); + sorted.sort(); + for (name, value) in &sorted { + field(&mut mac, name); + field(&mut mac, value); + } + mac + } + + /// `origin` is how devices reach this server, e.g. `https://sync.example.com`. + pub fn url( + &self, + origin: &str, + method: &str, + object_key: &str, + expires_ms: i64, + metadata: &[(String, String)], + ) -> String { + let expires = expires_ms.to_string(); + let signature = Self::mac(&self.keys[0], method, object_key, &expires, metadata) + .finalize() + .into_bytes(); + let mut query = url::form_urlencoded::Serializer::new(String::new()); + query.append_pair("key", object_key); + query.append_pair("expires", &expires); + for (name, value) in metadata { + query.append_pair(&format!("{META_PARAM}{name}"), value); + } + query.append_pair("sig", &hex::encode(signature)); + format!("{origin}{PATH}?{}", query.finish()) + } + + pub fn verify(&self, method: &str, query: &str, now_ms: i64) -> Result { + let mut key = None; + let mut expires = None; + let mut signature = None; + let mut metadata = Vec::new(); + for (name, value) in url::form_urlencoded::parse(query.as_bytes()) { + match name.as_ref() { + "key" if key.is_none() => key = Some(value.into_owned()), + "expires" if expires.is_none() => expires = Some(value.into_owned()), + "sig" if signature.is_none() => signature = Some(value.into_owned()), + other => match other.strip_prefix(META_PARAM) { + Some(meta) => metadata.push((meta.to_string(), value.into_owned())), + None => return Err(Refusal::BadSignature), + }, + } + } + let (Some(key), Some(expires), Some(signature)) = (key, expires, signature) else { + return Err(Refusal::BadSignature); + }; + let signature = hex::decode(signature).map_err(|_| Refusal::BadSignature)?; + let signed = self.keys.iter().any(|k| { + Self::mac(k, method, &key, &expires, &metadata) + .verify_slice(&signature) + .is_ok() + }); + if !signed { + return Err(Refusal::BadSignature); + } + let expires_ms: i64 = expires.parse().map_err(|_| Refusal::BadSignature)?; + if now_ms > expires_ms { + return Err(Refusal::Expired); + } + Ok(Grant { key, metadata }) + } +} + +/// An S3-style error document, so a device logs what S3 would have said. +fn s3_error(status: u16, code: &str, message: &str) -> Response { + let body = format!( + "\n{code}{message}" + ); + Response::builder() + .status(status) + .header(header::CONTENT_TYPE, "application/xml") + .header(header::CONTENT_LENGTH, body.len()) + .body(full(body)) + .expect("static response parts are valid") +} + +fn store_refusal(error: StoreError) -> Response { + match error { + StoreError::NotFound => s3_error(404, "NoSuchKey", "The specified key does not exist."), + StoreError::Service { status, code, .. } if (400..500).contains(&status) => { + s3_error(status, &code, "") + } + other => { + crate::log::error(PATH, &format!("Transfer failed err=\"{other}\"")); + s3_error( + 500, + "InternalError", + "We encountered an internal error. Please try again.", + ) + } + } +} + +fn header_text(parts: &Parts, name: impl header::AsHeaderName) -> Option { + parts + .headers + .get(name) + .map(|v| v.as_bytes().iter().map(|&b| b as char).collect()) +} + +pub async fn serve( + store: &Arc, + signer: &TransferSigner, + parts: &Parts, + body: Incoming, +) -> Response { + let method = parts.method.as_str(); + let grant = match signer.verify(method, parts.uri.query().unwrap_or(""), time::now_ms()) { + Ok(grant) if parts.method == Method::GET || parts.method == Method::PUT => grant, + Ok(_) | Err(Refusal::BadSignature) => { + drain(body).await; + return s3_error( + 403, + "SignatureDoesNotMatch", + "The request signature we calculated does not match the signature you provided.", + ); + } + Err(Refusal::Expired) => { + drain(body).await; + return s3_error(403, "AccessDenied", "Request has expired"); + } + }; + let slot = signer + .slots + .clone() + .acquire_owned() + .await + .expect("the semaphore is never closed"); + if parts.method == Method::GET { + drain(body).await; + return download(store, &grant.key, slot).await; + } + let response = upload(store, grant, parts, body).await; + drop(slot); + response +} + +async fn download( + store: &Arc, + key: &str, + slot: tokio::sync::OwnedSemaphorePermit, +) -> Response { + let object = match store.get_object(key).await { + Ok(object) => object, + Err(e) => return store_refusal(e), + }; + let mut response = Response::builder().status(200); + if let Some(length) = object.head.content_length { + response = response.header(header::CONTENT_LENGTH, length); + } + if let Some(content_type) = &object.head.content_type { + response = response.header(header::CONTENT_TYPE, content_type.as_str()); + } + if let Some(ms) = object.head.last_modified_ms { + response = response.header(header::LAST_MODIFIED, time::http_date(ms)); + } + for (name, value) in &object.head.metadata { + response = response.header(format!("x-amz-meta-{name}"), value.as_str()); + } + let key = key.to_string(); + // The slot is held until the body is sent. + let body = object.body.map(move |chunk| { + let _slot = &slot; + chunk.map(Frame::data).map_err(|e| { + crate::log::error(PATH, &format!("Download cut off key={key} err=\"{e}\"")); + BoxError::from(e) + }) + }); + response + .body(StreamBody::new(body).boxed_unsync()) + .unwrap_or_else(|_| { + s3_error( + 500, + "InternalError", + "Stored metadata is not a header value.", + ) + }) +} + +async fn upload( + store: &Arc, + grant: Grant, + parts: &Parts, + body: Incoming, +) -> Response { + let mut sent: Vec<(String, String)> = parts + .headers + .iter() + .filter_map(|(name, value)| { + name.as_str().strip_prefix("x-amz-meta-").map(|n| { + ( + n.to_string(), + value.as_bytes().iter().map(|&b| b as char).collect(), + ) + }) + }) + .collect(); + sent.sort(); + let mut signed = grant.metadata; + signed.sort(); + if sent != signed { + drain(body).await; + return s3_error( + 403, + "AccessDenied", + "There were headers present in the request which were not signed", + ); + } + let Some(length) = header_text(parts, header::CONTENT_LENGTH).and_then(|v| v.trim().parse().ok()) + else { + drain(body).await; + return s3_error( + 411, + "MissingContentLength", + "You must provide the Content-Length HTTP header.", + ); + }; + let content_type = header_text(parts, header::CONTENT_TYPE); + let stream = body + .into_data_stream() + .map(|chunk: Result| { + chunk.map_err(|e| StoreError::Transport(e.to_string())) + }); + match store + .put_object_stream( + &grant.key, + Box::pin(stream), + Upload { + length, + content_type: content_type.as_deref(), + metadata: &signed, + if_match: None, + }, + ) + .await + { + Ok(()) => Response::builder() + .status(200) + .header(header::CONTENT_LENGTH, 0) + .body(full(Bytes::new())) + .expect("static response parts are valid"), + Err(e) => store_refusal(e), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn signer() -> TransferSigner { + TransferSigner::new(&Keyring::parse("k1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=").unwrap()) + } + + fn query(url: &str) -> &str { + url.split_once('?').unwrap().1 + } + + #[test] + fn urls_verify_for_their_method_key_metadata_and_lifetime() { + let signer = signer(); + let metadata = vec![ + ("updated-at".to_string(), "17".to_string()), + ("group-id".to_string(), "g 1&x".to_string()), + ]; + let url = signer.url( + "https://sync.example.com", + "PUT", + "users/u/a b+c", + 1000, + &metadata, + ); + assert!(url.starts_with("https://sync.example.com/v1/transfer?key=users%2Fu%2Fa+b%2Bc&")); + let grant = signer.verify("PUT", query(&url), 1000).unwrap(); + assert_eq!(grant.key, "users/u/a b+c"); + assert_eq!(grant.metadata.len(), 2); + assert_eq!( + signer.verify("PUT", query(&url), 1001), + Err(Refusal::Expired) + ); + assert_eq!( + signer.verify("GET", query(&url), 0), + Err(Refusal::BadSignature) + ); + let tampered = query(&url).replace("updated-at=17", "updated-at=18"); + assert_eq!( + signer.verify("PUT", &tampered, 0), + Err(Refusal::BadSignature) + ); + let moved = query(&url).replace("key=users%2Fu", "key=users%2Fv"); + assert_eq!(signer.verify("PUT", &moved, 0), Err(Refusal::BadSignature)); + let added = format!("{}&meta-x=1", query(&url)); + assert_eq!(signer.verify("PUT", &added, 0), Err(Refusal::BadSignature)); + let extended = query(&url).replace("expires=1000", "expires=9999"); + assert_eq!( + signer.verify("PUT", &extended, 0), + Err(Refusal::BadSignature) + ); + } + + #[test] + fn an_old_key_still_verifies_after_rotation() { + let old = signer(); + let url = old.url("http://h", "GET", "k", 50, &[]); + let rotated = TransferSigner::new( + &Keyring::parse( + "k2:AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE=,k1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=", + ) + .unwrap(), + ); + assert!(rotated.verify("GET", query(&url), 0).is_ok()); + let removed = TransferSigner::new( + &Keyring::parse("k2:AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE=").unwrap(), + ); + assert_eq!( + removed.verify("GET", query(&url), 0), + Err(Refusal::BadSignature) + ); + } +} diff --git a/donut-sync/tests/at_rest.rs b/donut-sync/tests/at_rest.rs new file mode 100644 index 00000000..df95f3bf --- /dev/null +++ b/donut-sync/tests/at_rest.rs @@ -0,0 +1,452 @@ +//! Encryption at rest end to end: the binary with `SYNC_AT_REST_KEY` against +//! the compose S3 server, devices that speak only the S3 presigned protocol, +//! plaintext objects from before, and the migration command. + +mod common; + +use common::{SYNC_TOKEN, Server, bucket, http, post, raw_store, run_command, storage_env}; +use donut_sync::store::{Store, Upload, collect, once}; +use serde_json::{Value, json}; + +const K1: &str = "k1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="; +const K2: &str = "k2:AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE="; + +fn sample(len: usize) -> Vec { + (0..len).map(|i| (i * 131 + 17) as u8).collect() +} + +async fn sealed_server(name: &str, keys: &str) -> (Server, String) { + let bucket = bucket(name); + ( + Server::start(&bucket, &[("SYNC_AT_REST_KEY", Some(keys))]).await, + bucket, + ) +} + +async fn call(server: &Server, route: &str, body: Value) -> (u16, Value) { + post( + &http(), + &format!("{}/v1/objects/{route}", server.url), + Some(SYNC_TOKEN), + body, + ) + .await +} + +/// What the app does: presign, then PUT with the echoed metadata. +async fn upload(server: &Server, key: &str, body: Vec, updated_at: Option<&str>) -> String { + let mut request = json!({ "key": key, "contentType": "application/octet-stream" }); + if let Some(updated_at) = updated_at { + request["metadata"] = json!({ "updated-at": updated_at }); + } + let (status, presign) = call(server, "presign-upload", request).await; + assert_eq!(status, 200, "{presign}"); + let url = presign["url"].as_str().unwrap().to_string(); + let mut put = http() + .put(&url) + .header("Content-Length", body.len().to_string()) + .header("Content-Type", "application/octet-stream") + .body(body); + if let Some(metadata) = presign["metadata"].as_object() { + for (name, value) in metadata { + put = put.header(format!("x-amz-meta-{name}"), value.as_str().unwrap()); + } + } + let response = put.send().await.unwrap(); + assert_eq!(response.status(), 200, "{}", response.text().await.unwrap()); + url +} + +async fn download(server: &Server, key: &str) -> (u16, Vec) { + let (status, presign) = call(server, "presign-download", json!({ "key": key })).await; + assert_eq!(status, 200, "{presign}"); + let response = http() + .get(presign["url"].as_str().unwrap()) + .send() + .await + .unwrap(); + let status = response.status().as_u16(); + (status, response.bytes().await.unwrap().to_vec()) +} + +#[tokio::test] +async fn transfers_go_through_the_server_and_are_sealed_in_storage() { + let (server, bucket) = sealed_server("seal", K1).await; + let raw = raw_store(&bucket); + for (key, size) in [ + ("profiles/p/files/big", 3 * 65_536 + 77), + ("profiles/p/files/frame", 65_536), + ("profiles/p/files/empty", 0), + ] { + let plain = sample(size); + let url = upload(&server, key, plain.clone(), Some("1700000000")).await; + assert!( + url.starts_with(&format!("{}/v1/transfer?", server.url)), + "{url}" + ); + + let stored = collect(raw.get_object(key).await.unwrap().body) + .await + .unwrap(); + assert_eq!(&stored[..4], b"DSAR"); + assert_ne!(stored, plain); + let head = raw.head_object(key).await.unwrap(); + assert_eq!(head.metadata_value("donut-at-rest"), Some("1:k1")); + + let (status, stat) = call(&server, "stat", json!({ "key": key })).await; + assert_eq!(status, 200); + assert_eq!(stat["size"], json!(size)); + assert_eq!(stat["metadata"], json!({ "updated-at": "1700000000" })); + + assert_eq!(download(&server, key).await, (200, plain)); + } + let (_, readyz) = ( + 0, + http() + .get(format!("{}/readyz", server.url)) + .send() + .await + .unwrap() + .json::() + .await + .unwrap(), + ); + assert_eq!(readyz, json!({ "status": "ready", "s3": true })); + assert!( + server + .logs() + .contains("Encryption at rest on key=k1 keys=1") + ); +} + +#[tokio::test] +async fn batches_and_downloads_keep_the_s3_shapes() { + let (server, _) = sealed_server("batch", K1).await; + let (status, batch) = call( + &server, + "presign-upload-batch", + json!({ "items": [{ "key": "a" }, { "key": "b" }] }), + ) + .await; + assert_eq!(status, 200); + for item in batch["items"].as_array().unwrap() { + let key = item["key"].as_str().unwrap(); + assert!(item["expiresAt"].is_string()); + let response = http() + .put(item["url"].as_str().unwrap()) + .header("Content-Length", key.len().to_string()) + .body(key.to_string()) + .send() + .await + .unwrap(); + assert_eq!(response.status(), 200); + } + let (_, batch) = call( + &server, + "presign-download-batch", + json!({ "keys": ["a", "b"] }), + ) + .await; + for item in batch["items"].as_array().unwrap() { + let response = http() + .get(item["url"].as_str().unwrap()) + .send() + .await + .unwrap(); + assert_eq!( + response.text().await.unwrap(), + item["key"].as_str().unwrap() + ); + } + let (_, list) = call(&server, "list", json!({ "prefix": "" })).await; + let keys: Vec<&str> = list["objects"] + .as_array() + .unwrap() + .iter() + .map(|o| o["key"].as_str().unwrap()) + .collect(); + assert_eq!(keys, ["a", "b"]); +} + +#[tokio::test] +async fn transfer_urls_refuse_what_s3_would_refuse() { + let (server, _) = sealed_server("refuse", K1).await; + let client = http(); + let (_, presign) = call( + &server, + "presign-upload", + json!({ "key": "k", "metadata": { "updated-at": "5" } }), + ) + .await; + let url = presign["url"].as_str().unwrap(); + let put = |url: String| client.put(url).header("Content-Length", "1").body("x"); + + // The signed metadata must be sent, exactly. + let response = put(url.to_string()).send().await.unwrap(); + assert_eq!(response.status(), 403); + let response = put(url.to_string()) + .header("x-amz-meta-updated-at", "6") + .send() + .await + .unwrap(); + assert_eq!(response.status(), 403); + let response = put(url.to_string()) + .header("x-amz-meta-updated-at", "5") + .header("x-amz-meta-group-id", "g") + .send() + .await + .unwrap(); + assert_eq!(response.status(), 403); + // A changed URL, or the wrong method. + let response = put(url.replace("updated-at=5", "updated-at=6")) + .header("x-amz-meta-updated-at", "6") + .send() + .await + .unwrap(); + assert_eq!(response.status(), 403); + assert!( + response + .text() + .await + .unwrap() + .contains("SignatureDoesNotMatch") + ); + let response = client.get(url).send().await.unwrap(); + assert_eq!(response.status(), 403); + // No length, as S3 answers a chunked PUT. + let chunked = reqwest::Body::wrap_stream(futures_util::stream::iter([Ok::<_, std::io::Error>( + bytes::Bytes::from_static(b"x"), + )])); + let response = client + .put(url) + .header("x-amz-meta-updated-at", "5") + .body(chunked) + .send() + .await + .unwrap(); + assert_eq!(response.status(), 411); + // Expired. + let (_, short) = call( + &server, + "presign-download", + json!({ "key": "k", "expiresIn": 0.001 }), + ) + .await; + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + let response = client + .get(short["url"].as_str().unwrap()) + .send() + .await + .unwrap(); + assert_eq!(response.status(), 403); + assert!( + response + .text() + .await + .unwrap() + .contains("Request has expired") + ); + // Missing object. + assert_eq!(download(&server, "missing").await.0, 404); + // The right request still works. + let response = put(url.to_string()) + .header("x-amz-meta-updated-at", "5") + .send() + .await + .unwrap(); + assert_eq!(response.status(), 200); +} + +#[tokio::test] +async fn without_the_key_nothing_changes() { + let server = Server::start(&bucket("plain"), &[]).await; + let (_, presign) = call(&server, "presign-upload", json!({ "key": "k" })).await; + assert!( + presign["url"] + .as_str() + .unwrap() + .starts_with("http://127.0.0.1:8987/") + ); + let response = http() + .get(format!("{}/v1/transfer?key=k", server.url)) + .send() + .await + .unwrap(); + assert_eq!(response.status(), 404); + assert_eq!( + response.json::().await.unwrap()["message"], + "Cannot GET /v1/transfer?key=k" + ); + assert!(!server.logs().contains("Encryption at rest")); +} + +/// Writes `objects` straight to storage, as an older server left them. +async fn seed_plaintext(bucket: &str, objects: &[(&str, Vec)]) { + let raw = raw_store(bucket); + raw.create_bucket().await.ok(); + for (key, body) in objects { + raw + .put_object_stream( + key, + once(body.clone()), + Upload { + length: body.len() as u64, + content_type: Some("application/json"), + metadata: &[("updated-at".to_string(), "9".to_string())], + if_match: None, + }, + ) + .await + .unwrap(); + } +} + +#[tokio::test] +async fn plaintext_and_sealed_objects_read_side_by_side() { + let bucket = bucket("mixed"); + let old = sample(100_000); + seed_plaintext(&bucket, &[("proxies/old.json", old.clone())]).await; + let server = Server::start(&bucket, &[("SYNC_AT_REST_KEY", Some(K1))]).await; + upload(&server, "proxies/new.json", b"new".to_vec(), Some("10")).await; + + assert_eq!(download(&server, "proxies/old.json").await, (200, old)); + assert_eq!( + download(&server, "proxies/new.json").await, + (200, b"new".to_vec()) + ); + let (_, stat) = call(&server, "stat", json!({ "key": "proxies/old.json" })).await; + assert_eq!(stat["size"], json!(100_000)); + assert_eq!(stat["metadata"], json!({ "updated-at": "9" })); + + // Restore points copy sealed objects and read their sealed records. + upload(&server, "profiles/p/manifest.json", b"{}".to_vec(), None).await; + upload(&server, "profiles/p/files/a", b"A".to_vec(), None).await; + let (status, snapshot) = call( + &server, + "snapshots/create", + json!({ "profileRoot": "profiles/p/" }), + ) + .await; + assert_eq!(status, 200, "{snapshot}"); + let (_, listed) = call( + &server, + "snapshots/list", + json!({ "profileRoot": "profiles/p/" }), + ) + .await; + assert_eq!(listed["snapshots"][0]["id"], snapshot["id"]); + upload(&server, "profiles/p/files/a", b"changed".to_vec(), None).await; + let (status, _) = call( + &server, + "snapshots/restore-files", + json!({ "profileRoot": "profiles/p/", "id": snapshot["id"] }), + ) + .await; + assert_eq!(status, 200); + assert_eq!( + download(&server, "profiles/p/files/a").await, + (200, b"A".to_vec()) + ); +} + +#[tokio::test] +async fn migration_seals_everything_in_place_and_a_rerun_does_nothing() { + let bucket = bucket("migrate"); + let objects: Vec<(String, Vec)> = (0..25) + .map(|i| { + ( + format!("users/u/profiles/p/files/{i:02}"), + sample(i * 9_000), + ) + }) + .collect(); + let refs: Vec<(&str, Vec)> = objects + .iter() + .map(|(k, v)| (k.as_str(), v.clone())) + .collect(); + seed_plaintext(&bucket, &refs).await; + let storage = storage_env(&bucket); + let mut env: Vec<(&str, &str)> = storage.iter().map(|(k, v)| (*k, v.as_str())).collect(); + env.push(("SYNC_AT_REST_KEY", K1)); + let total: usize = objects.iter().map(|(_, v)| v.len()).sum(); + + let (code, out) = run_command(&["migrate-at-rest", "--dry-run"], &env); + assert_eq!(code, 0, "{out}"); + assert!( + out.contains(&format!( + "scanned=25 current=0 sealed=25 sealed_bytes={total} " + )), + "{out}" + ); + let raw = raw_store(&bucket); + let first = collect(raw.get_object(&objects[3].0).await.unwrap().body) + .await + .unwrap(); + assert_eq!(first, objects[3].1); + + let (code, out) = run_command( + &[ + "migrate-at-rest", + "--concurrency", + "3", + "--max-mib-per-sec", + "64", + ], + &env, + ); + assert_eq!(code, 0, "{out}"); + assert!(out.contains("conditional_put=false"), "{out}"); + assert!(out.contains("sealed=25"), "{out}"); + for (key, _) in &objects { + let head = raw.head_object(key).await.unwrap(); + assert_eq!(head.metadata_value("donut-at-rest"), Some("1:k1"), "{key}"); + assert_eq!(head.metadata_value("updated-at"), Some("9"), "{key}"); + assert_eq!(head.content_type.as_deref(), Some("application/json")); + } + let listed = raw + .list_objects_v2(&donut_sync::store::ListInput::default()) + .await + .unwrap(); + assert_eq!(listed.contents.len(), 25, "no temporary copy is left"); + + let (code, out) = run_command(&["migrate-at-rest"], &env); + assert_eq!(code, 0, "{out}"); + assert!(out.contains("current=25 sealed=0"), "{out}"); + + // The server reads what the migration wrote. Self-hosted keys are bare. + let server = Server::start(&bucket, &[("SYNC_AT_REST_KEY", Some(K1))]).await; + for (key, body) in &objects { + assert_eq!(download(&server, key).await, (200, body.clone())); + } + drop(server); + + // Rotation: k2 writes, k1 still reads, and the migration moves k1's objects. + let rotated = format!("{K2},{K1}"); + let server = Server::start(&bucket, &[("SYNC_AT_REST_KEY", Some(&rotated))]).await; + assert_eq!( + download(&server, &objects[5].0).await, + (200, objects[5].1.clone()) + ); + drop(server); + env.retain(|(k, _)| *k != "SYNC_AT_REST_KEY"); + env.push(("SYNC_AT_REST_KEY", &rotated)); + let (code, out) = run_command(&["migrate-at-rest"], &env); + assert_eq!(code, 0, "{out}"); + assert!(out.contains("rotated=25"), "{out}"); + let server = Server::start(&bucket, &[("SYNC_AT_REST_KEY", Some(K2))]).await; + assert_eq!( + download(&server, &objects[7].0).await, + (200, objects[7].1.clone()) + ); +} + +#[tokio::test] +async fn a_bad_key_stops_the_server_at_start() { + let storage = storage_env(&bucket("badkey")); + let mut env: Vec<(&str, &str)> = storage.iter().map(|(k, v)| (*k, v.as_str())).collect(); + env.push(("SYNC_TOKEN", SYNC_TOKEN)); + env.push(("SYNC_AT_REST_KEY", "k1:AAAA")); + let (code, out) = common::run_to_exit(&env); + assert_eq!(code, 1); + assert!(out.contains("must be 32 bytes"), "{out}"); +} diff --git a/donut-sync/tests/common/mod.rs b/donut-sync/tests/common/mod.rs index acaca042..1ca0c8c4 100644 --- a/donut-sync/tests/common/mod.rs +++ b/donut-sync/tests/common/mod.rs @@ -15,6 +15,29 @@ use donut_sync::store::{Store, StoreError}; pub const S3_ENDPOINT: &str = "http://127.0.0.1:8987"; pub const SYNC_TOKEN: &str = "test-sync-token-0123456789abcdef"; +/// The test storage, without the server: what is really stored. +pub fn raw_store(bucket: &str) -> S3Store { + S3Store::new( + Location::parse(S3_ENDPOINT, bucket, true).unwrap(), + Credentials { + access_key_id: "minioadmin".to_string(), + secret_access_key: "minioadmin".to_string(), + }, + "us-east-1".to_string(), + ) +} + +/// The storage settings the server takes, for a command run to its exit. +pub fn storage_env(bucket: &str) -> Vec<(&'static str, String)> { + vec![ + ("S3_ENDPOINT", S3_ENDPOINT.to_string()), + ("S3_ACCESS_KEY_ID", "minioadmin".to_string()), + ("S3_SECRET_ACCESS_KEY", "minioadmin".to_string()), + ("S3_BUCKET", bucket.to_string()), + ("S3_FORCE_PATH_STYLE", "true".to_string()), + ] +} + pub fn http() -> reqwest::Client { let _ = rustls::crypto::ring::default_provider().install_default(); reqwest::Client::builder().no_proxy().build().unwrap() @@ -78,7 +101,7 @@ impl Server { } } let output = Arc::new(Mutex::new(String::new())); - let mut child = spawn(&env); + let mut child = spawn(&[], &env); collect(&mut child, &output); let url = format!("http://127.0.0.1:{port}"); let client = http(); @@ -114,11 +137,12 @@ impl Drop for Server { } } -fn spawn(env: &[(String, String)]) -> Child { +fn spawn(args: &[&str], env: &[(String, String)]) -> Child { // A directory without a `.env`, so a developer's file cannot leak in. let dir = std::env::temp_dir().join("donut-sync-test-cwd"); std::fs::create_dir_all(&dir).unwrap(); Command::new(env!("CARGO_BIN_EXE_donut-sync")) + .args(args) .env_clear() .envs(env.iter().map(|(k, v)| (k.as_str(), v.as_str()))) .current_dir(dir) @@ -149,11 +173,16 @@ fn collect(child: &mut Child, output: &Arc>) { /// Runs the binary to its exit (a refused configuration) and returns the /// exit code and what it printed. pub fn run_to_exit(env: &[(&str, &str)]) -> (i32, String) { + run_command(&[], env) +} + +/// Runs the binary with `args` to its exit. +pub fn run_command(args: &[&str], env: &[(&str, &str)]) -> (i32, String) { let env: Vec<(String, String)> = env .iter() .map(|(k, v)| (k.to_string(), v.to_string())) .collect(); - let child = spawn(&env); + let child = spawn(args, &env); let out = child.wait_with_output().unwrap(); ( out.status.code().unwrap_or(-1),