diff --git a/donut-sync/.env.example b/donut-sync/.env.example index 717c7592..2bce6b3c 100644 --- a/donut-sync/.env.example +++ b/donut-sync/.env.example @@ -35,8 +35,10 @@ S3_FORCE_PATH_STYLE=true # 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. +# when E2E encryption is off in Donut Browser. Downloads and uploads from +# current apps then go through this server; older apps still upload straight to +# storage, and the server encrypts those files minutes later. 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. diff --git a/donut-sync/README.md b/donut-sync/README.md index 51ab1fec..d73ecb96 100644 --- a/donut-sync/README.md +++ b/donut-sync/README.md @@ -20,8 +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 +With `SYNC_AT_REST_KEY` set, the server encrypts every object it stores. +Downloads and uploads from current apps go through it, large files in parts. +Older apps still upload straight to storage, and the server encrypts those +files minutes later. Objects stored before stay readable; `donut-sync migrate-at-rest --help` encrypts them in place, and `donut-sync verify-at-rest` checks that none is left in plaintext. diff --git a/donut-sync/src/at_rest.rs b/donut-sync/src/at_rest.rs index c3b9b4e3..2facd04f 100644 --- a/donut-sync/src/at_rest.rs +++ b/donut-sync/src/at_rest.rs @@ -203,25 +203,53 @@ pub struct Sealer { pending: Vec, } +pub type Salt = [u8; SALT]; + +pub fn new_salt() -> Salt { + let mut salt = [0u8; SALT]; + rand::fill(&mut salt); + salt +} + +fn header_of(key_id: &str, salt: &Salt) -> Vec { + let mut header = Vec::with_capacity(header_len(key_id) as usize); + header.extend_from_slice(MAGIC); + header.push(VERSION); + header.push(key_id.len() as u8); + header.extend_from_slice(key_id.as_bytes()); + header.extend_from_slice(salt); + header +} + impl Sealer { + fn with(key: &Key, salt: &Salt, first_frame: u64) -> Self { + Self { + key: object_key(&key.secret, salt), + header: header_of(&key.id, salt), + index: first_frame, + pending: Vec::new(), + } + } + /// 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)) + let sealer = Self::with(keyring.current(), &new_salt(), 0); + let header = Bytes::from(sealer.header.clone()); + (sealer, header) + } + + /// One part of an object sealed in parts under `key_id` and `salt`: its + /// frames are numbered from `first_frame`, and only part 1 carries the + /// header. `None` when the keyring does not hold `key_id`. + pub fn for_part( + keyring: &Keyring, + key_id: &str, + salt: &Salt, + first_frame: u64, + ) -> Option<(Self, Option)> { + let sealer = Self::with(keyring.find(key_id)?, salt, first_frame); + let header = (first_frame == 0).then(|| Bytes::from(sealer.header.clone())); + Some((sealer, header)) } fn seal(&mut self, mut frame: Vec, last: bool) -> Bytes { @@ -247,6 +275,28 @@ impl Sealer { let frame = std::mem::take(&mut self.pending); self.seal(frame, true) } + + /// The end of a part: the object's final frame on the last part, else the + /// part's own last full frame. + fn finish_part(mut self, last: bool) -> Option { + if last { + return Some(self.finish()); + } + let frame = std::mem::take(&mut self.pending); + (!frame.is_empty()).then(|| self.seal(frame, false)) + } +} + +/// The stored size of one part of `plain` bytes. A part before the last holds +/// whole frames only. +pub fn sealed_part_len(key_id: &str, plain: u64, first: bool, last: bool) -> u64 { + let header = if first { header_len(key_id) } else { 0 }; + let frames = if last { + frames(plain) + } else { + plain / FRAME as u64 + }; + header + plain + frames * TAG as u64 } /// Decrypts a sealed byte stream, frame by frame. @@ -332,14 +382,15 @@ impl Opener { } enum Codec { - Seal(Option, Option), + /// The sealer, the header still to send, and whether this ends the object. + Seal(Option, Option, bool), Open(Option), } impl Codec { fn push(&mut self, data: &[u8], out: &mut VecDeque) -> Result<(), StoreError> { match self { - Codec::Seal(sealer, header) => { + Codec::Seal(sealer, header, _) => { if let Some(header) = header.take() { out.push_back(header); } @@ -352,11 +403,11 @@ impl Codec { fn finish(&mut self, out: &mut VecDeque) -> Result<(), StoreError> { match self { - Codec::Seal(sealer, header) => { + Codec::Seal(sealer, header, last) => { if let Some(header) = header.take() { out.push_back(header); } - out.push_back(sealer.take().expect("finished once").finish()); + out.extend(sealer.take().expect("finished once").finish_part(*last)); Ok(()) } Codec::Open(opener) => { @@ -414,7 +465,16 @@ fn transform(input: ByteStream, codec: Codec) -> ByteStream { /// `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))) + transform(input, Codec::Seal(Some(sealer), Some(header), true)) +} + +/// One part of an object sealed in parts (see `Sealer::for_part`). +pub fn seal_part_stream( + (sealer, header): (Sealer, Option), + last: bool, + input: ByteStream, +) -> ByteStream { + transform(input, Codec::Seal(Some(sealer), header, last)) } /// `input`, a sealed object, opened. A forged, cut or wrongly keyed object diff --git a/donut-sync/src/at_rest_tests.rs b/donut-sync/src/at_rest_tests.rs index a48adadc..7cb4d7a1 100644 --- a/donut-sync/src/at_rest_tests.rs +++ b/donut-sync/src/at_rest_tests.rs @@ -385,3 +385,64 @@ mod sealed_store { assert_eq!(store.get_object_text("b").await.unwrap(), "x"); } } + +#[tokio::test] +async fn an_object_sealed_in_parts_reads_like_one_sealed_whole() { + let keys = keyring(K1); + let part = 2 * FRAME; + for size in [ + 0, + 1, + part - 1, + part, + part + 1, + 2 * part, + 3 * part + FRAME + 5, + ] { + let plain = sample(size); + let salt = new_salt(); + let parts = size.div_ceil(part).max(1); + let mut sealed = Vec::new(); + for i in 0..parts { + let start = i * part; + let chunk = plain[start..(start + part).min(size)].to_vec(); + let last = i + 1 == parts; + let sealer = Sealer::for_part(&keys, "k1", &salt, (start / FRAME) as u64).unwrap(); + let len = chunk.len() as u64; + let out = collect(seal_part_stream(sealer, last, chunked(chunk, &[777]))) + .await + .unwrap(); + assert_eq!( + out.len() as u64, + sealed_part_len("k1", len, i == 0, last), + "size {size} part {i}" + ); + sealed.extend(out); + } + assert_eq!( + sealed.len() as u64, + sealed_len("k1", size as u64), + "size {size}" + ); + assert_eq!(open(keys.clone(), &sealed).unwrap(), plain, "size {size}"); + } + assert!(Sealer::for_part(&keys, "k9", &new_salt(), 0).is_none()); +} + +#[tokio::test] +async fn parts_out_of_place_do_not_open() { + let keys = keyring(K1); + let salt = new_salt(); + let part = |first: u64, last: bool, data: Vec| { + let sealer = Sealer::for_part(&keys, "k1", &salt, first).unwrap(); + collect(seal_part_stream(sealer, last, chunked(data, &[FRAME]))) + }; + let one = part(0, false, sample(FRAME)).await.unwrap(); + let two = part(1, true, sample(10)).await.unwrap(); + assert!(open(keys.clone(), &[one.clone(), two.clone()].concat()).is_ok()); + // A part numbered as if it came later, or a non-final part last. + let skipped = part(2, true, sample(10)).await.unwrap(); + assert!(open(keys.clone(), &[one.clone(), skipped].concat()).is_err()); + let unfinished = part(1, false, sample(FRAME)).await.unwrap(); + assert!(open(keys.clone(), &[one, unfinished].concat()).is_err()); +} diff --git a/donut-sync/src/http.rs b/donut-sync/src/http.rs index ff713bc5..e17271a7 100644 --- a/donut-sync/src/http.rs +++ b/donut-sync/src/http.rs @@ -39,7 +39,7 @@ pub fn full(bytes: impl Into) -> ResBody { .boxed_unsync() } -fn json_response(status: u16, body: &Value) -> Response { +pub(crate) fn json_response(status: u16, body: &Value) -> Response { let text = body.to_string(); Response::builder() .status(status) diff --git a/donut-sync/src/lib.rs b/donut-sync/src/lib.rs index f956914e..0a1dd815 100644 --- a/donut-sync/src/lib.rs +++ b/donut-sync/src/lib.rs @@ -18,12 +18,14 @@ pub mod migrate; pub mod net; pub mod s3; pub mod sealed_store; +pub mod sealer; pub mod sigv4; pub mod store; pub mod subscribe; pub mod sync; pub mod time; pub mod transfer; +pub mod transfer_parts; pub mod verify; use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr}; @@ -167,11 +169,11 @@ pub async fn build_state(config: &Config) -> Result, String> { ), ); ( - Arc::new(SealedStore::new(s3, keyring.clone())), + Arc::new(SealedStore::new(s3.clone(), keyring.clone())), Some(TransferSigner::new(keyring)), ) } - None => (s3, None), + None => (s3.clone(), None), }; let presigner = Presigner::new( Location::parse(&public_endpoint, &storage.bucket, storage.force_path_style)?, @@ -202,6 +204,9 @@ pub async fn build_state(config: &Config) -> Result, String> { sync.ensure_bucket_exists().await?; sync.warn_if_presign_endpoint_is_server_only(); + if let Some(keyring) = keyring { + crate::sealer::spawn(s3, keyring); + } Ok(Arc::new(AppState { guard, diff --git a/donut-sync/src/memory_store.rs b/donut-sync/src/memory_store.rs index cbbdb6ec..de934d14 100644 --- a/donut-sync/src/memory_store.rs +++ b/donut-sync/src/memory_store.rs @@ -9,8 +9,8 @@ use async_trait::async_trait; use md5::{Digest, Md5}; use crate::store::{ - ByteStream, Download, HeadOutput, ListInput, ListObject, ListOutput, Store, StoreError, Upload, - collect, once, + ByteStream, Download, HeadOutput, ListInput, ListObject, ListOutput, PendingUpload, Store, + StoreError, Upload, collect, once, }; pub type HeadHook = Box Result + Send + Sync>; @@ -24,6 +24,17 @@ pub struct Object { pub modified_ms: i64, } +/// S3's smallest part, except the last. +pub const MIN_PART: usize = 5 * 1024 * 1024; + +pub struct Multipart { + key: String, + content_type: Option, + metadata: Vec<(String, String)>, + parts: BTreeMap>, + initiated_ms: i64, +} + #[derive(Default)] pub struct MemoryStore { pub objects: Mutex>, @@ -32,6 +43,7 @@ pub struct MemoryStore { pub clock: AtomicI64, pub head_hook: Option, pub list_hook: Option, + pub uploads: Mutex>, } fn etag(body: &[u8]) -> String { @@ -268,4 +280,110 @@ impl Store for MemoryStore { ); Ok(()) } + + async fn create_multipart( + &self, + key: &str, + content_type: Option<&str>, + metadata: &[(String, String)], + ) -> Result { + self.call(); + let id = uuid::Uuid::new_v4().simple().to_string(); + self.uploads.lock().unwrap().insert( + id.clone(), + Multipart { + key: key.to_string(), + content_type: content_type.map(str::to_string), + metadata: metadata.to_vec(), + parts: BTreeMap::new(), + initiated_ms: self.tick(), + }, + ); + Ok(id) + } + + async fn upload_part( + &self, + key: &str, + upload_id: &str, + part: u32, + body: ByteStream, + length: u64, + ) -> Result { + self.call(); + let body = collect(body).await?; + if body.len() as u64 != length { + return Err(StoreError::Transport("length mismatch".to_string())); + } + let mut uploads = self.uploads.lock().unwrap(); + let upload = uploads + .get_mut(upload_id) + .filter(|u| u.key == key) + .ok_or(StoreError::NotFound)?; + let tag = etag(&body); + upload.parts.insert(part, body); + Ok(tag) + } + + async fn complete_multipart( + &self, + key: &str, + upload_id: &str, + etags: &[String], + ) -> Result<(), StoreError> { + self.call(); + let upload = self + .uploads + .lock() + .unwrap() + .remove(upload_id) + .filter(|u| u.key == key) + .ok_or(StoreError::NotFound)?; + let mut body = Vec::new(); + for (i, expected) in etags.iter().enumerate() { + let part = upload + .parts + .get(&(i as u32 + 1)) + .filter(|part| etag(part) == *expected) + .ok_or_else(|| StoreError::Service { + status: 400, + code: "InvalidPart".to_string(), + message: String::new(), + })?; + if i + 1 < etags.len() && part.len() < MIN_PART { + return Err(StoreError::Service { + status: 400, + code: "EntityTooSmall".to_string(), + message: String::new(), + }); + } + body.extend_from_slice(part); + } + self.insert(key, body, upload.metadata, upload.content_type); + Ok(()) + } + + async fn abort_multipart(&self, _key: &str, upload_id: &str) -> Result<(), StoreError> { + self.call(); + self.uploads.lock().unwrap().remove(upload_id); + Ok(()) + } + + async fn list_multipart(&self, prefix: &str) -> Result, StoreError> { + self.call(); + Ok( + self + .uploads + .lock() + .unwrap() + .iter() + .filter(|(_, u)| u.key.starts_with(prefix)) + .map(|(id, u)| PendingUpload { + key: u.key.clone(), + upload_id: id.clone(), + initiated_ms: Some(u.initiated_ms), + }) + .collect(), + ) + } } diff --git a/donut-sync/src/migrate.rs b/donut-sync/src/migrate.rs index 1c0815e6..67d7a567 100644 --- a/donut-sync/src/migrate.rs +++ b/donut-sync/src/migrate.rs @@ -29,7 +29,7 @@ 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/"; +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; @@ -148,7 +148,7 @@ pub struct Report { pub failed: u64, } -enum Outcome { +pub(crate) enum Outcome { Current, Sealed(u64), Rotated(u64), @@ -191,7 +191,7 @@ fn throttled(body: ByteStream, limit: Option>) -> ByteStream { } } -struct Migration { +pub(crate) struct Migration { store: Arc, keyring: Arc, dry_run: bool, @@ -199,6 +199,19 @@ struct Migration { limit: Option>, } +impl Migration { + /// Seals in place, as the migration does, without a rate limit. + pub(crate) fn sealing(store: Arc, keyring: Arc, conditional: bool) -> Self { + Self { + store, + keyring, + dry_run: false, + conditional, + limit: None, + } + } +} + fn precondition_failed(error: &StoreError) -> bool { matches!(error, StoreError::Service { status: 412, .. }) } @@ -234,7 +247,7 @@ pub async fn honors_if_match(store: &dyn Store) -> Result { impl Migration { /// `listed_ms` is the LastModified the listing gave: it has milliseconds, /// where a HEAD has seconds. - async fn object(&self, key: &str, listed_ms: Option) -> Outcome { + pub(crate) async fn object(&self, key: &str, listed_ms: Option) -> Outcome { let mut attempt = 0; loop { attempt += 1; @@ -432,7 +445,7 @@ pub async fn migrate( .contents .into_iter() .map(|object| (object.key, object.last_modified_ms)) - .filter(|(key, _)| !key.starts_with(TMP_PREFIX)) + .filter(|(key, _)| !key.starts_with(crate::sealer::INTERNAL_ROOT)) .collect(); let keys: Vec<&String> = objects.iter().map(|(key, _)| key).collect(); let outcomes: Vec<(String, Outcome)> = futures_util::stream::iter(objects.iter()) diff --git a/donut-sync/src/s3.rs b/donut-sync/src/s3.rs index fd023d46..f51aca02 100644 --- a/donut-sync/src/s3.rs +++ b/donut-sync/src/s3.rs @@ -11,7 +11,8 @@ use reqwest::Method; use crate::jsval::js_number; use crate::sigv4::{self, Credentials}; use crate::store::{ - ByteStream, Download, HeadOutput, ListInput, ListObject, ListOutput, Store, StoreError, Upload, + ByteStream, Download, HeadOutput, ListInput, ListObject, ListOutput, PendingUpload, Store, + StoreError, Upload, }; use crate::time; @@ -329,6 +330,38 @@ fn restore_controls(text: &str) -> String { .collect() } +type UploadsPage = (Vec, Option<(String, String)>); + +/// A ListMultipartUploads reply: the uploads, and where the next page starts. +fn parse_uploads(body: &[u8]) -> Result { + let text = std::str::from_utf8(body) + .map_err(|e| StoreError::Transport(format!("ListMultipartUploads reply is not UTF-8: {e}")))?; + let doc = roxmltree::Document::parse(text) + .map_err(|e| StoreError::Transport(format!("ListMultipartUploads reply is not XML: {e}")))?; + let root = doc.root_element(); + let mut uploads = Vec::new(); + let mut truncated = false; + for node in root.children().filter(|n| n.is_element()) { + match node.tag_name().name() { + "Upload" => uploads.push(PendingUpload { + key: child_text(node, "Key").unwrap_or("").to_string(), + upload_id: child_text(node, "UploadId").unwrap_or("").to_string(), + initiated_ms: child_text(node, "Initiated").and_then(time::parse_rfc3339_ms), + }), + "IsTruncated" => truncated = node.text().unwrap_or("").trim() == "true", + _ => {} + } + } + let next = if truncated { + child_text(root, "NextKeyMarker") + .zip(child_text(root, "NextUploadIdMarker")) + .map(|(key, id)| (key.to_string(), id.to_string())) + } else { + None + }; + Ok((uploads, next)) +} + fn parse_list(body: &[u8]) -> Result { let text = std::str::from_utf8(body) .map_err(|e| StoreError::Transport(format!("ListObjectsV2 reply is not UTF-8: {e}")))?; @@ -748,6 +781,150 @@ impl Store for S3Store { }) } + async fn create_multipart( + &self, + key: &str, + content_type: Option<&str>, + metadata: &[(String, String)], + ) -> Result { + let path = self.location.object_path(key); + let mut headers = Vec::new(); + if let Some(content_type) = content_type { + headers.push(("content-type".to_string(), content_type.to_string())); + } + for (name, value) in metadata { + headers.push((format!("x-amz-meta-{name}"), value.clone())); + } + let reply = self + .expect_success( + Method::POST, + &path, + &[("uploads".to_string(), String::new())], + &headers, + &[], + ) + .await?; + let text = String::from_utf8_lossy(&reply.body); + let doc = roxmltree::Document::parse(&text) + .map_err(|e| StoreError::Transport(format!("CreateMultipartUpload reply is not XML: {e}")))?; + child_text(doc.root_element(), "UploadId") + .filter(|id| !id.is_empty()) + .map(str::to_string) + .ok_or_else(|| { + StoreError::Transport("CreateMultipartUpload reply has no UploadId".to_string()) + }) + } + + /// One attempt: a stream cannot be sent twice. + async fn upload_part( + &self, + key: &str, + upload_id: &str, + part: u32, + body: ByteStream, + length: u64, + ) -> Result { + let path = self.location.object_path(key); + let query = [ + ("partNumber".to_string(), part.to_string()), + ("uploadId".to_string(), upload_id.to_string()), + ]; + let headers = [("content-length".to_string(), length.to_string())]; + let response = self + .request( + &Method::PUT, + &path, + &query, + &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); + } + head_of(response.headers()) + .etag + .ok_or_else(|| StoreError::Transport("UploadPart reply has no ETag".to_string())) + } + + async fn complete_multipart( + &self, + key: &str, + upload_id: &str, + etags: &[String], + ) -> Result<(), StoreError> { + let mut xml = String::from(""); + for (i, etag) in etags.iter().enumerate() { + xml.push_str(&format!( + "{}{}", + i + 1, + xml_escape(etag) + )); + } + xml.push_str(""); + let path = self.location.object_path(key); + let reply = self + .expect_success( + Method::POST, + &path, + &[("uploadId".to_string(), upload_id.to_string())], + &[("content-type".to_string(), "application/xml".to_string())], + xml.as_bytes(), + ) + .await?; + // Like a copy, a completion can fail with 200 and an error document. + if let Some((code, message)) = error_document(&reply.body) { + return Err(StoreError::Service { + status: reply.status, + code, + message, + }); + } + Ok(()) + } + + async fn abort_multipart(&self, key: &str, upload_id: &str) -> Result<(), StoreError> { + let path = self.location.object_path(key); + self + .expect_success( + Method::DELETE, + &path, + &[("uploadId".to_string(), upload_id.to_string())], + &[], + &[], + ) + .await + .map(|_| ()) + } + + async fn list_multipart(&self, prefix: &str) -> Result, StoreError> { + let path = self.location.bucket_path(); + let mut out = Vec::new(); + let mut marker: Option<(String, String)> = None; + loop { + let mut query = vec![ + ("uploads".to_string(), String::new()), + ("prefix".to_string(), prefix.to_string()), + ]; + if let Some((key, id)) = &marker { + query.push(("key-marker".to_string(), key.clone())); + query.push(("upload-id-marker".to_string(), id.clone())); + } + let reply = self + .expect_success(Method::GET, &path, &query, &[], &[]) + .await?; + let (uploads, next) = parse_uploads(&reply.body)?; + out.extend(uploads); + match next { + Some(next) if Some(&next) != marker.as_ref() => marker = Some(next), + _ => return Ok(out), + } + } + } + /// One attempt: a stream cannot be sent twice. async fn put_object_stream( &self, diff --git a/donut-sync/src/sealed_store.rs b/donut-sync/src/sealed_store.rs index 93752c8a..756698e5 100644 --- a/donut-sync/src/sealed_store.rs +++ b/donut-sync/src/sealed_store.rs @@ -12,7 +12,7 @@ use async_trait::async_trait; use crate::at_rest::{self, Keyring, MARKER}; use crate::store::{ - ByteStream, Download, HeadOutput, ListInput, ListOutput, Store, StoreError, Upload, + ByteStream, Download, HeadOutput, ListInput, ListOutput, PendingUpload, Store, StoreError, Upload, }; /// User metadata (`x-amz-meta-donut-orig-mtime`): the object's LastModified @@ -185,4 +185,50 @@ impl Store for SealedStore { ) .await } + + // Multipart parts pass through as they come: the caller (transfer.rs) + // seals each part as whole frames of one object. + + async fn create_multipart( + &self, + key: &str, + content_type: Option<&str>, + metadata: &[(String, String)], + ) -> Result { + self + .inner + .create_multipart(key, content_type, &self.with_marker(metadata)) + .await + } + + async fn upload_part( + &self, + key: &str, + upload_id: &str, + part: u32, + body: ByteStream, + length: u64, + ) -> Result { + self + .inner + .upload_part(key, upload_id, part, body, length) + .await + } + + async fn complete_multipart( + &self, + key: &str, + upload_id: &str, + etags: &[String], + ) -> Result<(), StoreError> { + self.inner.complete_multipart(key, upload_id, etags).await + } + + async fn abort_multipart(&self, key: &str, upload_id: &str) -> Result<(), StoreError> { + self.inner.abort_multipart(key, upload_id).await + } + + async fn list_multipart(&self, prefix: &str) -> Result, StoreError> { + self.inner.list_multipart(prefix).await + } } diff --git a/donut-sync/src/sealer.rs b/donut-sync/src/sealer.rs new file mode 100644 index 00000000..70317301 --- /dev/null +++ b/donut-sync/src/sealer.rs @@ -0,0 +1,444 @@ +//! Seals what older apps upload in plaintext. Older apps PUT to the direct +//! storage URL that presign-upload has always returned, because one request +//! through a proxy such as Cloudflare cannot carry a large file. The server +//! notes each key it hands such a URL for (`note_direct_uploads`), in the +//! bucket so a restart keeps the list, and seals those objects in place with +//! the migration's ETag-checked rewrite: on the first pass after the URL is +//! handed out (`PASS_EVERY`), and once more after the URL is dead. An hourly pass over everything written +//! since the last one catches what a lost record missed, and storage uploads +//! in parts that were never finished are aborted. + +use std::collections::HashSet; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use futures_util::StreamExt; +use serde_json::{Value, json}; + +use crate::at_rest::Keyring; +use crate::log; +use crate::migrate::{Migration, Outcome, honors_if_match}; +use crate::sealed_store::SealedStore; +use crate::store::{ListInput, Store, StoreError}; +use crate::time; + +/// This server's own objects. No client prefix matches it. +pub const INTERNAL_ROOT: &str = ".donut-sync-at-rest/"; +const PENDING: &str = ".donut-sync-at-rest/pending/"; +const STATE: &str = ".donut-sync-at-rest/state.json"; +const CONTEXT: &str = "Sealer"; +const PASS_EVERY: Duration = Duration::from_secs(120); +const BACKSTOP_EVERY: Duration = Duration::from_secs(3600); +/// Overlap between backstop passes, for clock skew between server and storage. +const BACKSTOP_OVERLAP_MS: i64 = 10 * 60 * 1000; +/// A first backstop pass looks back this far; the migration covers the rest. +const FIRST_BACKSTOP_MS: i64 = 2 * 3600 * 1000; +/// A record is kept this long after its URLs expire, for uploads in flight. +const RECORD_GRACE_MS: i64 = 5 * 60 * 1000; +/// Unfinished part uploads older than this are aborted: past any upload token. +const ABANDONED_MS: i64 = 26 * 3600 * 1000; +const CONCURRENCY: usize = 4; + +/// Notes `keys`, which devices may write in plaintext through direct storage +/// URLs until `expires_ms`. `store` is the sealed store: the record is sealed. +pub async fn note_direct_uploads( + store: &dyn Store, + keys: &[String], + expires_ms: i64, +) -> Result<(), StoreError> { + let record = format!( + "{PENDING}{expires_ms:013}-{}", + uuid::Uuid::new_v4().simple() + ); + store + .put_object( + &record, + json!({ "keys": keys }).to_string().into_bytes(), + "application/json", + &[], + ) + .await +} + +fn record_expiry(record: &str) -> Option { + record + .strip_prefix(PENDING)? + .split('-') + .next()? + .parse() + .ok() +} + +#[derive(Default, Debug, PartialEq)] +pub struct PassReport { + pub records: u64, + pub records_done: u64, + pub checked: u64, + pub sealed: u64, + pub failed: u64, +} + +impl PassReport { + fn add(&mut self, outcome: &Outcome) { + self.checked += 1; + match outcome { + Outcome::Sealed(_) | Outcome::Rotated(_) => self.sealed += 1, + Outcome::Failed(_) => self.failed += 1, + Outcome::Current | Outcome::Changed => {} + } + } +} + +pub struct Sealer { + raw: Arc, + sealed: SealedStore, + migration: Migration, + /// Records sealed once while their URLs live. A restart forgets them, and + /// they are sealed once more: harmless. + first_round: Mutex>, +} + +impl Sealer { + pub async fn new(raw: Arc, keyring: Arc) -> Result { + let conditional = honors_if_match(raw.as_ref()).await?; + Ok(Self { + sealed: SealedStore::new(raw.clone(), keyring.clone()), + migration: Migration::sealing(raw.clone(), keyring, conditional), + raw, + first_round: Mutex::new(HashSet::new()), + }) + } + + async fn seal_all(&self, keys: Vec<(String, Option)>, report: &mut PassReport) { + let outcomes: Vec<(String, Outcome)> = futures_util::stream::iter(keys) + .map(|(key, listed)| async move { + let outcome = self.migration.object(&key, listed).await; + (key, outcome) + }) + .buffer_unordered(CONCURRENCY) + .collect() + .await; + for (key, outcome) in outcomes { + if let Outcome::Failed(err) = &outcome { + log::warn( + CONTEXT, + &format!("Object not sealed key={key} err=\"{err}\""), + ); + } + report.add(&outcome); + } + } + + async fn list(&self, prefix: Option<&str>) -> Result)>, StoreError> { + let mut out = Vec::new(); + let mut token = None; + loop { + let page = self + .raw + .list_objects_v2(&ListInput { + prefix: prefix.map(str::to_string), + max_keys: Some("1000".to_string()), + continuation_token: token.clone(), + ..ListInput::default() + }) + .await?; + out.extend( + page + .contents + .into_iter() + .map(|object| (object.key, object.last_modified_ms)), + ); + token = page.next_continuation_token.filter(|_| page.is_truncated); + if token.is_none() { + return Ok(out); + } + } + } + + /// Seals the keys of each new record, and of each record whose URLs are + /// dead, which it then drops. + pub async fn pending_pass(&self, now_ms: i64) -> Result { + let mut report = PassReport::default(); + for (record, _) in self.list(Some(PENDING)).await? { + report.records += 1; + let dead = record_expiry(&record).is_some_and(|expiry| now_ms > expiry + RECORD_GRACE_MS); + if !dead && !self.first_round.lock().unwrap().insert(record.clone()) { + continue; + } + let keys: Vec<(String, Option)> = match self.sealed.get_object_text(&record).await { + Ok(text) => serde_json::from_str::(&text) + .ok() + .and_then(|v| v["keys"].as_array().cloned()) + .unwrap_or_default() + .iter() + .filter_map(|k| k.as_str().map(|k| (k.to_string(), None))) + .collect(), + Err(StoreError::NotFound) => continue, + Err(e) => { + log::warn( + CONTEXT, + &format!("Record unreadable key={record} err=\"{e}\""), + ); + report.failed += 1; + continue; + } + }; + let failed_before = report.failed; + self.seal_all(keys, &mut report).await; + if dead && report.failed == failed_before { + self.raw.delete_object(&record).await?; + self.first_round.lock().unwrap().remove(&record); + report.records_done += 1; + } + } + Ok(report) + } + + /// Seals everything written since the last pass, and aborts abandoned + /// uploads in parts. + pub async fn backstop(&self, now_ms: i64) -> Result { + let since = match self.sealed.get_object_text(STATE).await { + Ok(text) => serde_json::from_str::(&text) + .ok() + .and_then(|v| v["backstopSince"].as_i64()), + Err(StoreError::NotFound) => None, + Err(e) => return Err(e), + } + .unwrap_or(now_ms - FIRST_BACKSTOP_MS); + let recent: Vec<(String, Option)> = self + .list(None) + .await? + .into_iter() + .filter(|(key, modified)| { + !key.starts_with(INTERNAL_ROOT) + && modified.is_none_or(|ms| ms >= since - BACKSTOP_OVERLAP_MS) + }) + .collect(); + let mut report = PassReport::default(); + self.seal_all(recent, &mut report).await; + if report.failed == 0 { + self + .sealed + .put_object( + STATE, + json!({ "backstopSince": now_ms }).to_string().into_bytes(), + "application/json", + &[], + ) + .await?; + } + match self.raw.list_multipart("").await { + Ok(uploads) => { + for upload in uploads { + if upload + .initiated_ms + .is_some_and(|ms| now_ms - ms > ABANDONED_MS) + { + match self + .raw + .abort_multipart(&upload.key, &upload.upload_id) + .await + { + Ok(()) => log::log( + CONTEXT, + &format!("Abandoned upload aborted key={}", upload.key), + ), + Err(e) => log::warn( + CONTEXT, + &format!( + "Abandoned upload not aborted key={} err=\"{e}\"", + upload.key + ), + ), + } + } + } + } + // Some stores (rclone) cannot list them, and keep none. + Err(StoreError::NotFound) => {} + Err(e) => log::warn( + CONTEXT, + &format!("Unfinished uploads not listed err=\"{e}\""), + ), + } + Ok(report) + } +} + +fn log_pass(name: &str, outcome: Result, started: Instant) { + match outcome { + Ok(report) if report.sealed > 0 || report.failed > 0 || report.records_done > 0 => { + let line = format!( + "{name} records={} records_done={} checked={} sealed={} failed={} elapsed_ms={}", + report.records, + report.records_done, + report.checked, + report.sealed, + report.failed, + started.elapsed().as_millis() + ); + if report.failed > 0 { + log::warn(CONTEXT, &line); + } else { + log::log(CONTEXT, &line); + } + } + Ok(_) => {} + Err(e) => log::warn(CONTEXT, &format!("{name} stopped err=\"{e}\"")), + } +} + +/// Runs the passes for as long as the server does. +pub fn spawn(raw: Arc, keyring: Arc) { + tokio::spawn(async move { + // The first pass waits too: starting writes a probe object, which + // should not race the server's first requests. + tokio::time::sleep(PASS_EVERY).await; + let sealer = loop { + match Sealer::new(raw.clone(), keyring.clone()).await { + Ok(sealer) => break sealer, + Err(e) => { + log::warn(CONTEXT, &format!("Sealer not started err=\"{e}\"")); + tokio::time::sleep(PASS_EVERY).await; + } + } + }; + let mut last_backstop: Option = None; + loop { + let started = Instant::now(); + log_pass( + "Pending pass", + sealer.pending_pass(time::now_ms()).await, + started, + ); + if last_backstop.is_none_or(|at| at.elapsed() >= BACKSTOP_EVERY) { + let started = Instant::now(); + log_pass( + "Backstop pass", + sealer.backstop(time::now_ms()).await, + started, + ); + last_backstop = Some(Instant::now()); + } + tokio::time::sleep(PASS_EVERY).await; + } + }); +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::memory_store::MemoryStore; + use crate::sealed_store::sealed_by; + + const K1: &str = "k1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="; + + async fn setup() -> (Arc, SealedStore, Sealer) { + let raw = Arc::new(MemoryStore::default()); + let keyring = Arc::new(Keyring::parse(K1).unwrap()); + let sealed = SealedStore::new(raw.clone(), keyring.clone()); + let sealer = Sealer::new(raw.clone(), keyring).await.unwrap(); + (raw, sealed, sealer) + } + + async fn is_sealed(raw: &MemoryStore, key: &str) -> bool { + sealed_by(&raw.head_object(key).await.unwrap()).is_some() + } + + #[tokio::test] + async fn records_seal_direct_uploads_until_their_urls_are_dead() { + let (raw, sealed, sealer) = setup().await; + note_direct_uploads(&sealed, &["a".to_string(), "b".to_string()], 10_000) + .await + .unwrap(); + let record = raw + .keys() + .into_iter() + .find(|k| k.starts_with(PENDING)) + .unwrap(); + assert!( + raw.get(&record).unwrap().starts_with("DSAR"), + "the record is sealed" + ); + + // "a" is uploaded, "b" not yet. + raw.set("a", "plain a"); + let report = sealer.pending_pass(5_000).await.unwrap(); + assert_eq!( + (report.records, report.sealed, report.records_done), + (1, 1, 0) + ); + assert!(is_sealed(&raw, "a").await); + assert_eq!(sealed.get_object_text("a").await.unwrap(), "plain a"); + + // A live record is not sealed again. + raw.set("b", "plain b"); + let report = sealer.pending_pass(6_000).await.unwrap(); + assert_eq!((report.records, report.checked), (1, 0)); + + // The URL writes again before it expires; the pass after it is dead + // seals that too. + raw.set("a", "plain a2"); + let report = sealer + .pending_pass(10_000 + RECORD_GRACE_MS + 1) + .await + .unwrap(); + assert_eq!((report.sealed, report.records_done), (2, 1)); + assert_eq!(sealed.get_object_text("a").await.unwrap(), "plain a2"); + assert!(raw.keys().iter().all(|k| !k.starts_with(PENDING))); + } + + #[tokio::test] + async fn the_backstop_seals_recent_plaintext_and_remembers_where_it_stopped() { + use std::sync::atomic::Ordering::SeqCst; + let (raw, sealed, sealer) = setup().await; + let start = 10 * 3600 * 1000; + raw.clock.store(start, SeqCst); + raw.set("old", "x"); + raw.clock.store(start + 3 * 3600 * 1000, SeqCst); + raw.set("new", "y"); + let first = raw.clock.load(SeqCst) + 1; + let report = sealer.backstop(first).await.unwrap(); + assert_eq!( + report.sealed, 1, + "a first pass looks back FIRST_BACKSTOP_MS" + ); + assert!(is_sealed(&raw, "new").await); + assert!( + !is_sealed(&raw, "old").await, + "older objects are the migration's" + ); + assert!( + sealed + .get_object_text(STATE) + .await + .unwrap() + .contains(&first.to_string()) + ); + + raw.set("later", "z"); + let report = sealer.backstop(raw.clock.load(SeqCst) + 1).await.unwrap(); + assert_eq!(report.sealed, 1); + assert!(is_sealed(&raw, "later").await); + assert!(!is_sealed(&raw, "old").await); + } + + #[tokio::test] + async fn abandoned_uploads_in_parts_are_aborted() { + let (raw, _, sealer) = setup().await; + raw.create_multipart("k", None, &[]).await.unwrap(); + sealer.backstop(1).await.unwrap(); + assert_eq!(raw.list_multipart("").await.unwrap().len(), 1); + sealer.backstop(ABANDONED_MS + 1_000_000).await.unwrap(); + assert!(raw.list_multipart("").await.unwrap().is_empty()); + } + + #[test] + fn records_name_their_expiry() { + assert_eq!( + record_expiry(&format!("{PENDING}0000000012345-abc")), + Some(12345) + ); + assert_eq!(record_expiry("other"), None); + } +} diff --git a/donut-sync/src/store.rs b/donut-sync/src/store.rs index 1b91c2bd..07acb26f 100644 --- a/donut-sync/src/store.rs +++ b/donut-sync/src/store.rs @@ -138,6 +138,40 @@ pub trait Store: Send + Sync { body: ByteStream, upload: Upload<'_>, ) -> Result<(), StoreError>; + /// Starts a multipart upload; the object gets `content_type` and `metadata`. + async fn create_multipart( + &self, + key: &str, + content_type: Option<&str>, + metadata: &[(String, String)], + ) -> Result; + /// One part, `length` bytes from `body`. Returns its ETag. + async fn upload_part( + &self, + key: &str, + upload_id: &str, + part: u32, + body: ByteStream, + length: u64, + ) -> Result; + /// Joins the parts, in order, by their ETags. + async fn complete_multipart( + &self, + key: &str, + upload_id: &str, + etags: &[String], + ) -> Result<(), StoreError>; + async fn abort_multipart(&self, key: &str, upload_id: &str) -> Result<(), StoreError>; + /// Unfinished uploads under `prefix`. + async fn list_multipart(&self, prefix: &str) -> Result, StoreError>; +} + +/// An unfinished multipart upload. +#[derive(Debug, Clone, PartialEq)] +pub struct PendingUpload { + pub key: String, + pub upload_id: String, + pub initiated_ms: Option, } pub struct Download { diff --git a/donut-sync/src/sync.rs b/donut-sync/src/sync.rs index 24f22dc1..7199ccf7 100644 --- a/donut-sync/src/sync.rs +++ b/donut-sync/src/sync.rs @@ -195,17 +195,36 @@ impl SyncService { self.transfer.as_ref() } - fn put_url( + /// `transfer` of a presign-upload answer when the store is sealed: the + /// signed URL newer apps upload through, in parts. Older apps keep the + /// direct storage URL, whose writes the sealer seals afterwards. + fn transfer_json( &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), + ) -> Option { + let signer = self.transfer.as_ref()?; + Some(json!({ + "url": signer.url(origin, "PUT", key, expires_ms(now, expires_in), metadata), + "partSize": crate::transfer_parts::PART_SIZE, + })) + } + + /// Notes the keys handed direct storage URLs, so the sealer seals what + /// older apps write there. + async fn note_direct_uploads(&self, keys: &[String], now: i64, expires_in: f64) { + if self.transfer.is_none() || keys.is_empty() { + return; + } + if let Err(e) = + crate::sealer::note_direct_uploads(self.store.as_ref(), keys, expires_ms(now, expires_in)) + .await + { + // The hourly backstop pass still finds these objects. + log::warn(CONTEXT, &format!("Direct uploads not noted err=\"{e}\"")); } } @@ -240,16 +259,13 @@ impl SyncService { /// publish its storage host. /// Withheld too when transfers come here: devices never reach storage. pub fn diagnostic_storage_endpoint(&self) -> Option<&str> { - (!self.cloud && self.transfer.is_none()).then_some(self.public_endpoint.as_str()) + (!self.cloud).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; }; @@ -508,13 +524,15 @@ 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.put_url( - origin, - require_key(&key)?, - expires_in, - metadata.as_deref().unwrap_or(&[]), - now, - ); + let scoped = require_key(&key)?; + let metadata_or_none = metadata.as_deref().unwrap_or(&[]); + let url = self + .presigner + .put_url(scoped, expires_in, metadata_or_none, now); + let transfer = self.transfer_json(origin, scoped, expires_in, metadata_or_none, now); + self + .note_direct_uploads(&[scoped.to_string()], now, expires_in) + .await; if ctx.mode == Mode::Cloud && field_starts_with(raw_key, "profiles/")? { self.report_profile_usage_async(ctx); @@ -527,6 +545,9 @@ impl SyncService { if let Some(metadata) = &metadata { out.insert("metadata".to_string(), metadata_json(metadata)); } + if let Some(transfer) = transfer { + out.insert("transfer".to_string(), transfer); + } Ok(Value::Object(out)) } @@ -632,7 +653,9 @@ impl SyncService { let objects: Vec = response .contents .iter() - .filter(|object| !object.key.ends_with(MANIFEST_KEY)) + .filter(|object| { + !object.key.ends_with(MANIFEST_KEY) && !object.key.starts_with(crate::sealer::INTERNAL_ROOT) + }) .map(|object| { let key = if !user_prefix.is_empty() { object.key.strip_prefix(user_prefix).unwrap_or(&object.key) @@ -708,16 +731,23 @@ impl SyncService { ) .await?; let mut out_items = Vec::with_capacity(keys.len()); + let mut direct = Vec::with_capacity(keys.len()); for (raw, key) in raw_keys.iter().zip(&keys) { - let url = self.put_url(origin, require_key(key)?, expires_in, &[], now); + let scoped = require_key(key)?; + let url = self.presigner.put_url(scoped, expires_in, &[], now); let mut item = Map::new(); if let Field::Value(value) = raw { item.insert("key".to_string(), (*value).clone()); } item.insert("url".to_string(), Value::String(url)); item.insert("expiresAt".to_string(), Value::String(expires_at.clone())); + if let Some(transfer) = self.transfer_json(origin, scoped, expires_in, &[], now) { + item.insert("transfer".to_string(), transfer); + } out_items.push(Value::Object(item)); + direct.push(scoped.to_string()); } + self.note_direct_uploads(&direct, now, expires_in).await; if ctx.mode == Mode::Cloud { let mut any_profile = false; diff --git a/donut-sync/src/transfer.rs b/donut-sync/src/transfer.rs index a7561d6a..57d4a078 100644 --- a/donut-sync/src/transfer.rs +++ b/donut-sync/src/transfer.rs @@ -3,6 +3,9 @@ //! 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-*`. +//! +//! A request body through a proxy such as Cloudflare has a size and time +//! limit, so newer apps upload large files in parts (`transfer_parts`). use std::sync::Arc; @@ -19,16 +22,23 @@ use crate::at_rest::Keyring; use crate::http::{BoxError, ResBody, drain, full}; use crate::store::{Store, StoreError, Upload}; use crate::time; +use crate::transfer_parts; 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-"; +/// Query parameters a part request adds to a signed URL. The upload token +/// authenticates them, so they are not part of the URL's signature. +const UPLOAD_PARAM: &str = "upload"; +const PART_PARAM: &str = "part"; +const OP_PARAM: &str = "op"; pub struct TransferSigner { /// The current key's first; any of them verifies. keys: Vec<[u8; 32]>, + pub(crate) keyring: Arc, slots: Arc, } @@ -39,6 +49,16 @@ pub struct Grant { pub metadata: Vec<(String, String)>, } +/// A URL whose signature holds, with what a part request added to it. +#[derive(Debug, PartialEq)] +pub struct Signed { + pub grant: Grant, + pub expires_ms: i64, + pub upload: Option, + pub part: Option, + pub op: Option, +} + #[derive(Debug, PartialEq)] pub enum Refusal { Expired, @@ -51,9 +71,10 @@ fn field(mac: &mut Hmac, value: &str) { } impl TransferSigner { - pub fn new(keyring: &Keyring) -> Self { + pub fn new(keyring: &Arc) -> Self { Self { keys: keyring.url_keys(), + keyring: keyring.clone(), slots: Arc::new(tokio::sync::Semaphore::new(MAX_TRANSFERS)), } } @@ -102,15 +123,28 @@ impl TransferSigner { } pub fn verify(&self, method: &str, query: &str, now_ms: i64) -> Result { + let signed = self.check(method, query)?; + if now_ms > signed.expires_ms { + return Err(Refusal::Expired); + } + Ok(signed.grant) + } + + /// The signature only; the caller decides whether the URL's time counts. + pub fn check(&self, method: &str, query: &str) -> Result { let mut key = None; let mut expires = None; let mut signature = None; let mut metadata = Vec::new(); + let (mut upload, mut part, mut op) = (None, None, None); 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()), + UPLOAD_PARAM if upload.is_none() => upload = Some(value.into_owned()), + PART_PARAM if part.is_none() => part = Some(value.into_owned()), + OP_PARAM if op.is_none() => op = 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), @@ -130,15 +164,34 @@ impl TransferSigner { 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 }) + Ok(Signed { + grant: Grant { key, metadata }, + expires_ms, + upload, + part, + op, + }) + } + + /// HMAC of `data` under the current key, for upload tokens. + pub(crate) fn tag(&self, data: &[u8]) -> Vec { + let mut mac = Hmac::::new_from_slice(&self.keys[0]).expect("HMAC takes any key length"); + mac.update(data); + mac.finalize().into_bytes().to_vec() + } + + /// Whether `tag` is `data`'s under any key. + pub(crate) fn tag_holds(&self, data: &[u8], tag: &[u8]) -> bool { + self.keys.iter().any(|key| { + let mut mac = Hmac::::new_from_slice(key).expect("HMAC takes any key length"); + mac.update(data); + mac.verify_slice(tag).is_ok() + }) } } /// An S3-style error document, so a device logs what S3 would have said. -fn s3_error(status: u16, code: &str, message: &str) -> Response { +pub(crate) fn s3_error(status: u16, code: &str, message: &str) -> Response { let body = format!( "\n{code}{message}" ); @@ -150,7 +203,7 @@ fn s3_error(status: u16, code: &str, message: &str) -> Response { .expect("static response parts are valid") } -fn store_refusal(error: StoreError) -> Response { +pub(crate) 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) => { @@ -167,7 +220,7 @@ fn store_refusal(error: StoreError) -> Response { } } -fn header_text(parts: &Parts, name: impl header::AsHeaderName) -> Option { +pub(crate) fn header_text(parts: &Parts, name: impl header::AsHeaderName) -> Option { parts .headers .get(name) @@ -180,37 +233,110 @@ pub async fn serve( 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) => { + // A GET URL reads; a PUT URL writes, in one request or in parts. + let signed_for = if parts.method == Method::GET { + "GET" + } else { + "PUT" + }; + let signed = match signer.check(signed_for, parts.uri.query().unwrap_or("")) { + Ok(signed) => signed, + Err(_) => { 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"); + return bad_signature(); } }; + // Parts carry an upload token, which outlives the URL that started them. + if signed.upload.is_none() && time::now_ms() > signed.expires_ms { + 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; + let response = match (&parts.method, signed.upload.is_some()) { + (&Method::GET, false) => { + drain(body).await; + return download(store, &signed.grant.key, slot).await; + } + (&Method::PUT, false) => upload(store, signed.grant, parts, body).await, + (&Method::POST, false) => transfer_parts::start(store, signer, signed.grant, parts, body).await, + (&Method::PUT | &Method::POST | &Method::DELETE, true) => { + transfer_parts::serve(store, signer, signed, parts, body).await + } + _ => { + drain(body).await; + bad_signature() + } + }; drop(slot); response } +pub(crate) fn bad_signature() -> Response { + s3_error( + 403, + "SignatureDoesNotMatch", + "The request signature we calculated does not match the signature you provided.", + ) +} + +/// The `x-amz-meta-*` a request sent, when they are exactly the signed ones. +pub(crate) fn signed_metadata( + parts: &Parts, + grant: Grant, +) -> Result, Box>> { + 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 { + return Err(Box::new(s3_error( + 403, + "AccessDenied", + "There were headers present in the request which were not signed", + ))); + } + Ok(signed) +} + +/// A request body as a store stream. +pub(crate) fn body_stream(body: Incoming) -> crate::store::ByteStream { + Box::pin( + body + .into_data_stream() + .map(|chunk: Result| { + chunk.map_err(|e| StoreError::Transport(e.to_string())) + }), + ) +} + +pub(crate) fn content_length(parts: &Parts) -> Option { + header_text(parts, header::CONTENT_LENGTH).and_then(|v| v.trim().parse().ok()) +} + +pub(crate) fn missing_length() -> Response { + s3_error( + 411, + "MissingContentLength", + "You must provide the Content-Length HTTP header.", + ) +} + async fn download( store: &Arc, key: &str, @@ -259,48 +385,23 @@ async fn upload( 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 { + let key = grant.key.clone(); + let signed = match signed_metadata(parts, grant) { + Ok(signed) => signed, + Err(refusal) => { + drain(body).await; + return *refusal; + } + }; + let Some(length) = content_length(parts) else { 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.", - ); + return missing_length(); }; 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), + &key, + body_stream(body), Upload { length, content_type: content_type.as_deref(), @@ -324,7 +425,9 @@ mod tests { use super::*; fn signer() -> TransferSigner { - TransferSigner::new(&Keyring::parse("k1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=").unwrap()) + TransferSigner::new(&Arc::new( + Keyring::parse("k1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=").unwrap(), + )) } fn query(url: &str) -> &str { @@ -377,16 +480,16 @@ mod tests { 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( + let rotated = TransferSigner::new(&Arc::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(), - ); + let removed = TransferSigner::new(&Arc::new( + Keyring::parse("k2:AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE=").unwrap(), + )); assert_eq!( removed.verify("GET", query(&url), 0), Err(Refusal::BadSignature) diff --git a/donut-sync/src/transfer_parts.rs b/donut-sync/src/transfer_parts.rs new file mode 100644 index 00000000..4683e193 --- /dev/null +++ b/donut-sync/src/transfer_parts.rs @@ -0,0 +1,357 @@ +//! Uploads in parts through this server, for a store sealed at rest. A proxy +//! in front of the server (Cloudflare) limits each request's size and time, +//! so a large file goes up in `PART_SIZE` pieces on the signed PUT URL that +//! presign-upload hands out as `transfer.url`: +//! +//! ```text +//! POST url x-donut-upload-length: N, the signed x-amz-meta-*, Content-Type +//! -> 200 {"upload": T, "partSize": P, "parts": n} +//! PUT url&upload=T&part=i i in 1..=n, exactly P bytes (the last part: the rest) +//! -> 200, ETag +//! POST url&upload=T&op=complete {"parts": [ETag of part 1, ...]} -> 200 +//! DELETE url&upload=T -> 204 +//! ``` +//! +//! The token T is signed and names the object, the storage upload and the +//! sealing salt, so the server keeps no state between requests. A part may be +//! sent again. Each part is sealed as whole frames of one object, numbered +//! where the part starts, so the joined object reads like any sealed object. + +use base64::Engine; +use bytes::Bytes; +use http_body_util::BodyExt; +use hyper::body::Incoming; +use hyper::http::request::Parts; +use hyper::{Method, Response, StatusCode, header}; +use serde::{Deserialize, Serialize}; +use std::sync::Arc; + +use crate::at_rest::{self, FRAME, Salt, Sealer}; +use crate::http::{ResBody, drain, full, json_response}; +use crate::store::Store; +use crate::time; +use crate::transfer::{ + Grant, Signed, TransferSigner, bad_signature, body_stream, content_length, header_text, + missing_length, s3_error, signed_metadata, store_refusal, +}; + +/// Plaintext bytes per part: far under a proxy's request limit, and sent +/// within its timeout on a slow uplink. A whole number of frames, and above +/// S3's 5 MB minimum part. +pub const PART_SIZE: u64 = 8 * 1024 * 1024; +/// How long an upload may take once started. +const UPLOAD_TTL_MS: i64 = 24 * 3600 * 1000; +pub const LENGTH_HEADER: &str = "x-donut-upload-length"; +/// S3's most parts in one upload. +const MAX_PARTS: u64 = 10_000; +const MAX_COMPLETE_BODY: usize = 1024 * 1024; +const TOKEN_CONTEXT: &[u8] = b"donut-sync upload token v1\n"; + +const _: () = assert!(PART_SIZE.is_multiple_of(FRAME as u64) && PART_SIZE >= 5 * 1024 * 1024); + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +struct Token { + key: String, + upload_id: String, + key_id: String, + salt: String, + length: u64, + part_size: u64, + expires_ms: i64, +} + +impl Token { + fn parts(&self) -> u64 { + self.length.div_ceil(self.part_size).max(1) + } + + fn part_len(&self, part: u64) -> u64 { + if part < self.parts() { + self.part_size + } else { + self.length - (self.parts() - 1) * self.part_size + } + } + + fn salt(&self) -> Option { + hex::decode(&self.salt).ok()?.try_into().ok() + } + + fn seal(&self, signer: &TransferSigner) -> String { + let json = serde_json::to_vec(self).expect("a token serializes"); + let tag = signer.tag(&[TOKEN_CONTEXT, &json].concat()); + format!( + "{}.{}", + base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(&json), + hex::encode(tag) + ) + } + + /// The token, when it is ours, for `key`, and not expired. + fn open(signer: &TransferSigner, text: &str, key: &str, now_ms: i64) -> Option { + let (body, tag) = text.split_once('.')?; + let json = base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(body) + .ok()?; + if !signer.tag_holds(&[TOKEN_CONTEXT, &json].concat(), &hex::decode(tag).ok()?) { + return None; + } + let token: Self = serde_json::from_slice(&json).ok()?; + (token.key == key && now_ms <= token.expires_ms).then_some(token) + } +} + +fn invalid(code: &str, message: &str) -> Response { + s3_error(400, code, message) +} + +/// Starts an upload of `x-donut-upload-length` bytes. +pub async fn start( + store: &Arc, + signer: &TransferSigner, + grant: Grant, + parts: &Parts, + body: Incoming, +) -> Response { + drain(body).await; + let key = grant.key.clone(); + let metadata = match signed_metadata(parts, grant) { + Ok(metadata) => metadata, + Err(refusal) => return *refusal, + }; + let Some(length) = header_text(parts, LENGTH_HEADER).and_then(|v| v.trim().parse::().ok()) + else { + return invalid( + "MissingUploadLength", + "Send the file size as x-donut-upload-length.", + ); + }; + if length.div_ceil(PART_SIZE) > MAX_PARTS { + return invalid( + "EntityTooLarge", + "The file is too large to upload in parts.", + ); + } + let content_type = header_text(parts, header::CONTENT_TYPE); + let upload_id = match store + .create_multipart(&key, content_type.as_deref(), &metadata) + .await + { + Ok(id) => id, + Err(e) => return store_refusal(e), + }; + let token = Token { + key, + upload_id, + key_id: signer.keyring.current().id().to_string(), + salt: hex::encode(at_rest::new_salt()), + length, + part_size: PART_SIZE, + expires_ms: time::now_ms() + UPLOAD_TTL_MS, + }; + json_response( + 200, + &serde_json::json!({ + "upload": token.seal(signer), + "partSize": token.part_size, + "parts": token.parts(), + }), + ) +} + +/// A part, the completion or the abort of a started upload. +pub async fn serve( + store: &Arc, + signer: &TransferSigner, + signed: Signed, + parts: &Parts, + body: Incoming, +) -> Response { + let token = signed + .upload + .as_deref() + .and_then(|text| Token::open(signer, text, &signed.grant.key, time::now_ms())); + let Some(token) = token else { + drain(body).await; + return bad_signature(); + }; + match (&parts.method, signed.op.as_deref()) { + (&Method::PUT, None) => part(store, signer, &token, signed.part.as_deref(), parts, body).await, + (&Method::POST, Some("complete")) => complete(store, &token, body).await, + (&Method::DELETE, None) => { + drain(body).await; + match store.abort_multipart(&token.key, &token.upload_id).await { + Ok(()) => Response::builder() + .status(StatusCode::NO_CONTENT) + .body(full(Bytes::new())) + .expect("static response parts are valid"), + Err(e) => store_refusal(e), + } + } + _ => { + drain(body).await; + bad_signature() + } + } +} + +async fn part( + store: &Arc, + signer: &TransferSigner, + token: &Token, + number: Option<&str>, + parts: &Parts, + body: Incoming, +) -> Response { + let Some(number) = number + .and_then(|n| n.parse::().ok()) + .filter(|n| (1..=token.parts()).contains(n)) + else { + drain(body).await; + return invalid( + "InvalidPart", + "The part number is not one of this upload's.", + ); + }; + let Some(length) = content_length(parts) else { + drain(body).await; + return missing_length(); + }; + let expected = token.part_len(number); + if length != expected { + drain(body).await; + return invalid( + "InvalidPart", + &format!("Part {number} must be {expected} bytes, not {length}."), + ); + } + let first_frame = (number - 1) * token.part_size / FRAME as u64; + let sealer = token + .salt() + .and_then(|salt| Sealer::for_part(&signer.keyring, &token.key_id, &salt, first_frame)); + let Some(sealer) = sealer else { + drain(body).await; + return invalid("InvalidUpload", "This upload's key is no longer held."); + }; + let last = number == token.parts(); + let sealed_len = at_rest::sealed_part_len(&token.key_id, length, number == 1, last); + let sealed = at_rest::seal_part_stream(sealer, last, body_stream(body)); + match store + .upload_part( + &token.key, + &token.upload_id, + number as u32, + sealed, + sealed_len, + ) + .await + { + Ok(etag) => Response::builder() + .status(200) + .header(header::ETAG, etag) + .header(header::CONTENT_LENGTH, 0) + .body(full(Bytes::new())) + .unwrap_or_else(|_| s3_error(500, "InternalError", "Storage gave an invalid ETag.")), + Err(e) => store_refusal(e), + } +} + +#[derive(Deserialize)] +struct Completion { + parts: Vec, +} + +async fn complete(store: &Arc, token: &Token, body: Incoming) -> Response { + let bytes = match http_body_util::Limited::new(body, MAX_COMPLETE_BODY) + .collect() + .await + { + Ok(collected) => collected.to_bytes(), + Err(_) => return invalid("MalformedXML", "The completion body is too large."), + }; + let Ok(completion) = serde_json::from_slice::(&bytes) else { + return invalid("MalformedXML", "Send {\"parts\": [ETag, ...]}."); + }; + if completion.parts.len() as u64 != token.parts() { + return invalid( + "InvalidPart", + &format!("This upload has {} parts.", token.parts()), + ); + } + match store + .complete_multipart(&token.key, &token.upload_id, &completion.parts) + .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::*; + use crate::at_rest::Keyring; + + fn signer(keys: &str) -> TransferSigner { + TransferSigner::new(&Arc::new(Keyring::parse(keys).unwrap())) + } + + const K1: &str = "k1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="; + + fn token(length: u64) -> Token { + Token { + key: "users/u/a".to_string(), + upload_id: "id".to_string(), + key_id: "k1".to_string(), + salt: hex::encode([7u8; 32]), + length, + part_size: PART_SIZE, + expires_ms: 1000, + } + } + + #[test] + fn tokens_are_bound_to_their_key_signer_and_time() { + let signer = signer(K1); + let text = token(5).seal(&signer); + assert_eq!( + Token::open(&signer, &text, "users/u/a", 1000), + Some(token(5)) + ); + assert_eq!(Token::open(&signer, &text, "users/u/b", 0), None); + assert_eq!(Token::open(&signer, &text, "users/u/a", 1001), None); + let other = signer_other(); + assert_eq!(Token::open(&other, &text, "users/u/a", 0), None); + let (body, tag) = text.split_once('.').unwrap(); + let mut json = base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(body) + .unwrap(); + let at = json.iter().position(|&b| b == b'5').unwrap(); + json[at] = b'6'; + let forged = format!( + "{}.{tag}", + base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(json) + ); + assert_eq!(Token::open(&signer, &forged, "users/u/a", 0), None); + } + + fn signer_other() -> TransferSigner { + signer("k2:AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE=") + } + + #[test] + fn parts_split_the_length() { + assert_eq!((token(0).parts(), token(0).part_len(1)), (1, 0)); + assert_eq!(token(PART_SIZE).parts(), 1); + let t = token(2 * PART_SIZE + 3); + assert_eq!(t.parts(), 3); + assert_eq!( + (t.part_len(1), t.part_len(2), t.part_len(3)), + (PART_SIZE, PART_SIZE, 3) + ); + } +} diff --git a/donut-sync/src/verify.rs b/donut-sync/src/verify.rs index 721b5eba..03bb0cbc 100644 --- a/donut-sync/src/verify.rs +++ b/donut-sync/src/verify.rs @@ -12,7 +12,6 @@ use futures_util::StreamExt; use crate::at_rest::{self, Keyring, MARKER}; use crate::config::Config; use crate::log; -use crate::migrate::TMP_PREFIX; use crate::sealed_store::sealed_by; use crate::store::{ListInput, Store, StoreError}; @@ -135,8 +134,9 @@ pub struct Report { pub unknown_key: u64, /// A marker of another format version, or a size no sealed object has. pub invalid: u64, - /// Leftover temporary copies of an interrupted migration. - pub temporary: u64, + /// This server's own objects: records of direct uploads, and leftovers of + /// an interrupted migration. + pub internal: u64, /// Gone between the listing and the HEAD. pub vanished: u64, pub sampled: u64, @@ -160,10 +160,10 @@ impl Report { )); } line.push_str(&format!( - " unknown_key={} invalid={} temporary={} vanished={} sampled={} sample_failed={}", + " unknown_key={} invalid={} internal={} vanished={} sampled={} sample_failed={}", self.unknown_key, self.invalid, - self.temporary, + self.internal, self.vanished, self.sampled, self.sample_failed @@ -252,12 +252,12 @@ pub async fn verify( }) .await .map_err(|e| format!("Listing failed: {e}"))?; - let (temporary, keys): (Vec, Vec) = page + let (internal, keys): (Vec, Vec) = page .contents .into_iter() .map(|object| object.key) - .partition(|key| key.starts_with(TMP_PREFIX)); - report.temporary += temporary.len() as u64; + .partition(|key| key.starts_with(crate::sealer::INTERNAL_ROOT)); + report.internal += internal.len() as u64; let classified: Vec<(String, Classified)> = futures_util::stream::iter(keys) .map(|key| { let store = store.clone(); @@ -394,7 +394,7 @@ mod tests { .put_object("c", b"ccc".to_vec(), "text/plain", &[]) .await .unwrap(); - raw.set(&format!("{TMP_PREFIX}left"), "x"); + raw.set(&format!("{}left", crate::migrate::TMP_PREFIX), "x"); let keys = keyring(&format!("{K1},{K2}")); let before = verify(raw.clone(), keys.clone(), &Options::default()) @@ -408,7 +408,7 @@ mod tests { assert_eq!(before.sealed["k2"].objects, 1); assert_eq!(before.sealed["k2"].plain_bytes, 3); assert_eq!( - (before.temporary, before.sampled, before.sample_failed), + (before.internal, before.sampled, before.sample_failed), (1, 1, 0) ); diff --git a/donut-sync/tests/at_rest.rs b/donut-sync/tests/at_rest.rs index 16512943..a8a17c39 100644 --- a/donut-sync/tests/at_rest.rs +++ b/donut-sync/tests/at_rest.rs @@ -33,7 +33,8 @@ async fn call(server: &Server, route: &str, body: Value) -> (u16, Value) { .await } -/// What the app does: presign, then PUT with the echoed metadata. +/// What a newer app does with a small file: presign, then PUT to the +/// transfer URL 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 { @@ -41,7 +42,15 @@ async fn upload(server: &Server, key: &str, body: Vec, updated_at: Option<&s } let (status, presign) = call(server, "presign-upload", request).await; assert_eq!(status, 200, "{presign}"); - let url = presign["url"].as_str().unwrap().to_string(); + // Older apps keep the direct storage URL. + assert!( + presign["url"] + .as_str() + .unwrap() + .starts_with(common::S3_ENDPOINT) + ); + assert_eq!(presign["transfer"]["partSize"], json!(8 * 1024 * 1024)); + let url = presign["transfer"]["url"].as_str().unwrap().to_string(); let mut put = http() .put(&url) .header("Content-Length", body.len().to_string()) @@ -111,7 +120,11 @@ async fn transfers_go_through_the_server_and_are_sealed_in_storage() { .await .unwrap(), ); - assert_eq!(readyz, json!({ "status": "ready", "s3": true })); + // Older apps still upload to storage directly, so its address is named. + assert_eq!( + readyz, + json!({ "status": "ready", "s3": true, "storageEndpoint": common::S3_ENDPOINT }) + ); assert!( server .logs() @@ -132,8 +145,13 @@ async fn batches_and_downloads_keep_the_s3_shapes() { for item in batch["items"].as_array().unwrap() { let key = item["key"].as_str().unwrap(); assert!(item["expiresAt"].is_string()); + // "a" as a newer app sends it, "b" as an older one. + let url = match key { + "a" => item["transfer"]["url"].as_str().unwrap(), + _ => item["url"].as_str().unwrap(), + }; let response = http() - .put(item["url"].as_str().unwrap()) + .put(url) .header("Content-Length", key.len().to_string()) .body(key.to_string()) .send() @@ -178,7 +196,7 @@ async fn transfer_urls_refuse_what_s3_would_refuse() { json!({ "key": "k", "metadata": { "updated-at": "5" } }), ) .await; - let url = presign["url"].as_str().unwrap(); + let url = presign["transfer"]["url"].as_str().unwrap(); let put = |url: String| client.put(url).header("Content-Length", "1").body("x"); // The signed metadata must be sent, exactly. @@ -538,3 +556,220 @@ async fn a_bad_key_stops_the_server_at_start() { assert_eq!(code, 1); assert!(out.contains("must be 32 bytes"), "{out}"); } + +const PART: usize = 8 * 1024 * 1024; + +/// Presign as a newer app does, and return (transfer URL, signed metadata). +async fn presign_transfer(server: &Server, key: &str) -> (String, Value) { + let (status, presign) = call( + server, + "presign-upload", + json!({ "key": key, "contentType": "application/octet-stream", "metadata": { "updated-at": "77" } }), + ) + .await; + assert_eq!(status, 200, "{presign}"); + ( + presign["transfer"]["url"].as_str().unwrap().to_string(), + presign["metadata"].clone(), + ) +} + +async fn start_parts(url: &str, length: usize) -> reqwest::Response { + http() + .post(url) + .header("x-donut-upload-length", length.to_string()) + .header("Content-Type", "application/octet-stream") + .header("x-amz-meta-updated-at", "77") + .send() + .await + .unwrap() +} + +fn part_url(url: &str, upload: &str, part: usize) -> String { + format!("{url}&upload={upload}&part={part}") +} + +#[tokio::test] +async fn large_files_upload_in_parts_into_one_sealed_object() { + let (server, bucket) = sealed_server("parts", K1).await; + let key = "profiles/p/files/large"; + let plain = sample(2 * PART + 12_345); + let (url, _) = presign_transfer(&server, key).await; + + let started = start_parts(&url, plain.len()).await; + assert_eq!(started.status(), 200); + let started: Value = started.json().await.unwrap(); + assert_eq!( + (started["partSize"].as_u64(), started["parts"].as_u64()), + (Some(PART as u64), Some(3)) + ); + let upload = started["upload"].as_str().unwrap(); + + let mut etags = Vec::new(); + for (i, chunk) in plain.chunks(PART).enumerate() { + let response = http() + .put(part_url(&url, upload, i + 1)) + .header("Content-Length", chunk.len().to_string()) + .body(chunk.to_vec()) + .send() + .await + .unwrap(); + assert_eq!(response.status(), 200, "{}", response.text().await.unwrap()); + etags.push(response.headers()["etag"].to_str().unwrap().to_string()); + } + // A part sent again replaces itself. + let again = http() + .put(part_url(&url, upload, 2)) + .header("Content-Length", PART.to_string()) + .body(plain[PART..2 * PART].to_vec()) + .send() + .await + .unwrap(); + etags[1] = again.headers()["etag"].to_str().unwrap().to_string(); + let done = http() + .post(format!("{url}&upload={upload}&op=complete")) + .json(&json!({ "parts": etags })) + .send() + .await + .unwrap(); + assert_eq!(done.status(), 200, "{}", done.text().await.unwrap()); + + let raw = raw_store(&bucket); + let head = raw.head_object(key).await.unwrap(); + assert_eq!(head.metadata_value("donut-at-rest"), Some("1:k1")); + let (_, stat) = call(&server, "stat", json!({ "key": key })).await; + assert_eq!(stat["size"], json!(plain.len())); + assert_eq!(stat["metadata"]["updated-at"], "77"); + assert_eq!(download(&server, key).await, (200, plain)); +} + +#[tokio::test] +async fn part_requests_refuse_what_does_not_fit_the_upload() { + let (server, _) = sealed_server("partrefuse", K1).await; + let (url, _) = presign_transfer(&server, "k").await; + // Unsigned metadata, and no length. + let response = http() + .post(&url) + .header("x-donut-upload-length", "10") + .send() + .await + .unwrap(); + assert_eq!(response.status(), 403); + let response = http() + .post(&url) + .header("x-amz-meta-updated-at", "77") + .send() + .await + .unwrap(); + assert_eq!(response.status(), 400); + + let started: Value = start_parts(&url, PART + 1).await.json().await.unwrap(); + let upload = started["upload"].as_str().unwrap().to_string(); + let put = |part: usize, len: usize| { + http() + .put(part_url(&url, &upload, part)) + .header("Content-Length", len.to_string()) + .body(vec![0u8; len]) + .send() + }; + assert_eq!( + put(1, PART - 1).await.unwrap().status(), + 400, + "a short part" + ); + assert_eq!(put(3, 1).await.unwrap().status(), 400, "no part 3"); + assert_eq!(put(0, 1).await.unwrap().status(), 400, "no part 0"); + // A forged token, or this token on another key's URL. + let forged = upload.replacen('.', ".0", 1); + let response = http() + .put(part_url(&url, &forged, 2)) + .header("Content-Length", "1") + .body(vec![0u8]) + .send() + .await + .unwrap(); + assert_eq!(response.status(), 403); + let (other, _) = presign_transfer(&server, "other").await; + let response = http() + .put(part_url(&other, &upload, 2)) + .header("Content-Length", "1") + .body(vec![0u8]) + .send() + .await + .unwrap(); + assert_eq!(response.status(), 403); + // The wrong number of parts on completion. + let one = put(2, 1).await.unwrap(); + let etag = one.headers()["etag"].to_str().unwrap().to_string(); + let done = http() + .post(format!("{url}&upload={upload}&op=complete")) + .json(&json!({ "parts": [etag] })) + .send() + .await + .unwrap(); + assert_eq!(done.status(), 400); + // Aborted, the upload takes no more parts. + let aborted = http() + .delete(format!("{url}&upload={upload}")) + .send() + .await + .unwrap(); + assert_eq!(aborted.status(), 204); + assert_eq!(download(&server, "k").await.0, 404); +} + +#[tokio::test] +async fn older_apps_upload_directly_and_the_sealer_seals_it() { + let (server, bucket) = sealed_server("direct", K1).await; + let key = "profiles/p/files/old-app"; + let (status, presign) = call( + &server, + "presign-upload", + json!({ "key": key, "metadata": { "updated-at": "5" } }), + ) + .await; + assert_eq!(status, 200); + let plain = sample(300_000); + let response = http() + .put(presign["url"].as_str().unwrap()) + .header("Content-Length", plain.len().to_string()) + .header("x-amz-meta-updated-at", "5") + .body(plain.clone()) + .send() + .await + .unwrap(); + assert_eq!(response.status(), 200); + let raw: std::sync::Arc = std::sync::Arc::new(raw_store(&bucket)); + assert!( + raw + .head_object(key) + .await + .unwrap() + .metadata_value("donut-at-rest") + .is_none() + ); + // Readable before it is sealed. + assert_eq!(download(&server, key).await, (200, plain.clone())); + + let keyring = std::sync::Arc::new(donut_sync::at_rest::Keyring::parse(K1).unwrap()); + let sealer = donut_sync::sealer::Sealer::new(raw.clone(), keyring) + .await + .unwrap(); + let report = sealer + .pending_pass(donut_sync::time::now_ms()) + .await + .unwrap(); + assert_eq!((report.records, report.sealed, report.failed), (1, 1, 0)); + let head = raw.head_object(key).await.unwrap(); + assert_eq!(head.metadata_value("donut-at-rest"), Some("1:k1")); + assert_eq!(head.metadata_value("updated-at"), Some("5")); + assert_eq!(download(&server, key).await, (200, plain)); + // The record stays until its URL is dead. + let (_, list) = call(&server, "list", json!({ "prefix": "" })).await; + assert_eq!(list["objects"].as_array().unwrap().len(), 1, "{list}"); + let report = sealer + .pending_pass(donut_sync::time::now_ms() + 2 * 3600 * 1000) + .await + .unwrap(); + assert_eq!(report.records_done, 1); +} diff --git a/src-tauri/src/sync/client.rs b/src-tauri/src/sync/client.rs index bd84351e..7010b3ea 100644 --- a/src-tauri/src/sync/client.rs +++ b/src-tauri/src/sync/client.rs @@ -23,6 +23,29 @@ const CONNECT_TIMEOUT: Duration = Duration::from_secs(8); /// only when nothing arrives for a full minute. const READ_TIMEOUT: Duration = Duration::from_secs(60); +/// Tries per part of an upload in parts. The file's own retries come on top. +const PART_ATTEMPTS: u32 = 3; +const PART_RETRY_DELAY: Duration = Duration::from_secs(1); +/// The size of a file uploaded in parts, which the server checks every part against. +const UPLOAD_LENGTH_HEADER: &str = "x-donut-upload-length"; + +/// The server's answer to the start of an upload in parts. +#[derive(serde::Deserialize)] +struct PartsStarted { + upload: String, + #[serde(rename = "partSize")] + part_size: u64, + parts: u64, +} + +/// `url` with `pairs` added to its query. +fn with_query(url: &str, pairs: &[(&str, &str)]) -> SyncResult { + let mut url = reqwest::Url::parse(url) + .map_err(|e| SyncError::NetworkError(format!("Invalid transfer URL: {e}")))?; + url.query_pairs_mut().extend_pairs(pairs); + Ok(url.into()) +} + #[derive(Clone)] pub struct SyncClient { client: Client, @@ -227,42 +250,201 @@ impl SyncClient { Ok(all_objects) } + /// Uploads to a presign answer: through `transfer` when the server offers + /// it, else to `presigned_url`. pub async fn upload_bytes( &self, presigned_url: &str, + transfer: Option<&SealedTransfer>, data: &[u8], content_type: Option<&str>, ) -> SyncResult<()> { self - .upload_bytes_with_metadata(presigned_url, data, content_type, None) + .upload_bytes_with_metadata(presigned_url, transfer, data, content_type, None) .await } - /// PUT to a presigned URL, sending `metadata` as `x-amz-meta-*` headers. These - /// MUST be exactly the metadata the presign signed (from - /// `PresignUploadResponse::metadata`) or S3 rejects the request. + /// `upload_bytes`, sending `metadata` as `x-amz-meta-*` headers. These MUST + /// be exactly the metadata the presign signed (from + /// `PresignUploadResponse::metadata`) or the upload is refused. pub async fn upload_bytes_with_metadata( + &self, + presigned_url: &str, + transfer: Option<&SealedTransfer>, + data: &[u8], + content_type: Option<&str>, + metadata: Option<&std::collections::HashMap>, + ) -> SyncResult<()> { + match transfer { + Some(transfer) if data.len() as u64 > transfer.part_size => { + self + .upload_in_parts(transfer, data, content_type, metadata) + .await + } + Some(transfer) => { + self + .put_once(&transfer.url, data, content_type, metadata) + .await + } + None => { + self + .put_once(presigned_url, data, content_type, metadata) + .await + } + } + } + + fn with_upload_headers( + mut req: reqwest::RequestBuilder, + content_type: Option<&str>, + metadata: Option<&std::collections::HashMap>, + ) -> reqwest::RequestBuilder { + if let Some(ct) = content_type { + req = req.header("Content-Type", ct); + } + for (k, v) in metadata.into_iter().flatten() { + req = req.header(format!("x-amz-meta-{k}"), v); + } + req + } + + /// Sends a request of an upload in parts; any answer but a success is an + /// error carrying the server's own words. + async fn send_part_request( + &self, + url: &str, + req: reqwest::RequestBuilder, + ) -> SyncResult { + let response = req + .send() + .await + .map_err(|e| SyncError::NetworkError(super::preflight::transport_reason_for(url, &e)))?; + if !response.status().is_success() { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + return Err(SyncError::NetworkError(format!( + "Upload failed with status {status}: {body}" + ))); + } + Ok(response) + } + + /// The server's protocol for files larger than one request may carry: + /// start with the size, send each part (retried on its own), then join + /// them. A failure aborts the upload, so storage keeps nothing half done. + async fn upload_in_parts( + &self, + transfer: &SealedTransfer, + data: &[u8], + content_type: Option<&str>, + metadata: Option<&std::collections::HashMap>, + ) -> SyncResult<()> { + let start = Self::with_upload_headers( + self + .client + .post(&transfer.url) + .header(UPLOAD_LENGTH_HEADER, data.len().to_string()) + .header("Content-Length", "0"), + content_type, + metadata, + ); + let started: PartsStarted = self + .send_part_request(&transfer.url, start) + .await? + .json() + .await + .map_err(|e| SyncError::SerializationError(e.to_string()))?; + let outcome = self.send_parts(transfer, &started, data).await; + if outcome.is_err() { + let abort = with_query(&transfer.url, &[("upload", &started.upload)])?; + if let Err(e) = self.client.delete(&abort).send().await { + log::debug!("Upload abort failed err=\"{e}\""); + } + } + outcome + } + + async fn send_parts( + &self, + transfer: &SealedTransfer, + started: &PartsStarted, + data: &[u8], + ) -> SyncResult<()> { + let part_size = usize::try_from(started.part_size) + .ok() + .filter(|size| *size > 0) + .ok_or_else(|| SyncError::InvalidData("The server gave no part size".to_string()))?; + let chunks: Vec<&[u8]> = data.chunks(part_size).collect(); + if chunks.len() as u64 != started.parts { + return Err(SyncError::InvalidData(format!( + "The server expects {} parts, the file has {}", + started.parts, + chunks.len() + ))); + } + let mut etags = Vec::with_capacity(chunks.len()); + for (index, chunk) in chunks.iter().enumerate() { + let number = (index + 1).to_string(); + let url = with_query( + &transfer.url, + &[("upload", &started.upload), ("part", &number)], + )?; + let mut attempt = 0; + let etag = loop { + attempt += 1; + let req = self + .client + .put(&url) + .header("Content-Length", chunk.len().to_string()) + .body(chunk.to_vec()); + match self.send_part_request(&url, req).await { + Ok(response) => { + break response + .headers() + .get("etag") + .and_then(|v| v.to_str().ok()) + .map(str::to_string) + .ok_or_else(|| { + SyncError::InvalidData(format!("Part {number} came back without an ETag")) + })?; + } + Err(e) if attempt >= PART_ATTEMPTS => return Err(e), + Err(e) => { + log::debug!("Part retry part={number} attempt={attempt} err=\"{e}\""); + tokio::time::sleep(PART_RETRY_DELAY * attempt).await; + } + } + }; + etags.push(etag); + } + let complete = with_query( + &transfer.url, + &[("upload", &started.upload), ("op", "complete")], + )?; + let req = self + .client + .post(&complete) + .json(&serde_json::json!({ "parts": etags })); + self.send_part_request(&complete, req).await.map(|_| ()) + } + + /// One PUT, as S3 takes it. + async fn put_once( &self, presigned_url: &str, data: &[u8], content_type: Option<&str>, metadata: Option<&std::collections::HashMap>, ) -> SyncResult<()> { - let mut req = self - .client - .put(presigned_url) - .header("Content-Length", data.len().to_string()) - .body(data.to_vec()); - - if let Some(ct) = content_type { - req = req.header("Content-Type", ct); - } - - if let Some(meta) = metadata { - for (k, v) in meta { - req = req.header(format!("x-amz-meta-{k}"), v); - } - } + let req = Self::with_upload_headers( + self + .client + .put(presigned_url) + .header("Content-Length", data.len().to_string()) + .body(data.to_vec()), + content_type, + metadata, + ); // The storage host here comes from the presigned URL, so on a self-hosted // server it is whatever the server signed against, frequently an address @@ -491,3 +673,200 @@ impl SyncClient { Ok(()) } } + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::HashMap; + use wiremock::matchers::{body_bytes, body_json, header, method, path, query_param}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + fn client() -> SyncClient { + SyncClient::new("http://127.0.0.1:1".to_string(), "unused".to_string()) + } + + fn transfer(server: &MockServer, part_size: u64) -> SealedTransfer { + SealedTransfer { + url: format!("{}/v1/transfer?key=k&expires=1&sig=s", server.uri()), + part_size, + } + } + + fn meta() -> HashMap { + HashMap::from([("updated-at".to_string(), "7".to_string())]) + } + + #[tokio::test] + async fn without_a_transfer_the_presigned_url_takes_the_file() { + let server = MockServer::start().await; + Mock::given(method("PUT")) + .and(path("/bucket/k")) + .and(body_bytes(b"data".to_vec())) + .respond_with(ResponseTemplate::new(200)) + .expect(1) + .mount(&server) + .await; + client() + .upload_bytes(&format!("{}/bucket/k", server.uri()), None, b"data", None) + .await + .unwrap(); + } + + #[tokio::test] + async fn a_file_within_one_part_goes_through_the_server_in_one_request() { + let server = MockServer::start().await; + Mock::given(method("PUT")) + .and(path("/v1/transfer")) + .and(header("x-amz-meta-updated-at", "7")) + .and(header("content-type", "application/json")) + .and(body_bytes(b"data".to_vec())) + .respond_with(ResponseTemplate::new(200)) + .expect(1) + .mount(&server) + .await; + client() + .upload_bytes_with_metadata( + "http://storage.invalid/never", + Some(&transfer(&server, 4)), + b"data", + Some("application/json"), + Some(&meta()), + ) + .await + .unwrap(); + } + + #[tokio::test] + async fn a_larger_file_goes_up_in_parts_and_a_failed_part_is_sent_again() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/v1/transfer")) + .and(header("x-donut-upload-length", "10")) + .and(header("x-amz-meta-updated-at", "7")) + .respond_with( + ResponseTemplate::new(200) + .set_body_json(serde_json::json!({ "upload": "T.1", "partSize": 4, "parts": 3 })), + ) + .expect(1) + .mount(&server) + .await; + for (part, body) in [("1", &b"0123"[..]), ("2", b"4567"), ("3", b"89")] { + if part == "2" { + Mock::given(method("PUT")) + .and(query_param("part", part)) + .respond_with(ResponseTemplate::new(503)) + .up_to_n_times(1) + .with_priority(1) + .mount(&server) + .await; + } + Mock::given(method("PUT")) + .and(query_param("upload", "T.1")) + .and(query_param("part", part)) + .and(query_param("key", "k")) + .and(body_bytes(body.to_vec())) + .respond_with(ResponseTemplate::new(200).insert_header("etag", format!("\"e{part}\""))) + .expect(1) + .mount(&server) + .await; + } + Mock::given(method("POST")) + .and(query_param("op", "complete")) + .and(query_param("upload", "T.1")) + .and(body_json( + serde_json::json!({ "parts": ["\"e1\"", "\"e2\"", "\"e3\""] }), + )) + .respond_with(ResponseTemplate::new(200)) + .expect(1) + .mount(&server) + .await; + client() + .upload_bytes_with_metadata( + "http://storage.invalid/never", + Some(&transfer(&server, 4)), + b"0123456789", + None, + Some(&meta()), + ) + .await + .unwrap(); + } + + #[tokio::test] + async fn a_part_that_keeps_failing_aborts_the_upload() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/v1/transfer")) + .respond_with( + ResponseTemplate::new(200) + .set_body_json(serde_json::json!({ "upload": "T.2", "partSize": 4, "parts": 2 })), + ) + .mount(&server) + .await; + Mock::given(method("PUT")) + .respond_with(ResponseTemplate::new(500).set_body_string("no")) + .expect(u64::from(PART_ATTEMPTS)) + .mount(&server) + .await; + Mock::given(method("DELETE")) + .and(query_param("upload", "T.2")) + .respond_with(ResponseTemplate::new(204)) + .expect(1) + .mount(&server) + .await; + let error = client() + .upload_bytes( + "http://storage.invalid/never", + Some(&transfer(&server, 4)), + b"01234567", + None, + ) + .await + .unwrap_err(); + assert!(error.to_string().contains("500"), "{error}"); + } + + #[tokio::test] + async fn a_part_count_that_disagrees_with_the_file_is_refused_before_sending() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with( + ResponseTemplate::new(200) + .set_body_json(serde_json::json!({ "upload": "T.3", "partSize": 4, "parts": 5 })), + ) + .mount(&server) + .await; + Mock::given(method("DELETE")) + .respond_with(ResponseTemplate::new(204)) + .expect(1) + .mount(&server) + .await; + assert!(client() + .upload_bytes( + "http://x.invalid", + Some(&transfer(&server, 4)), + b"0123456", + None + ) + .await + .is_err()); + } + + #[test] + fn answers_from_servers_that_do_not_seal_have_no_transfer() { + let old: PresignUploadResponse = + serde_json::from_str(r#"{"url":"u","expiresAt":"t"}"#).unwrap(); + assert_eq!(old.transfer, None); + let new: PresignUploadBatchItemResponse = serde_json::from_str( + r#"{"key":"k","url":"u","expiresAt":"t","transfer":{"url":"v","partSize":8388608}}"#, + ) + .unwrap(); + assert_eq!( + new.transfer, + Some(SealedTransfer { + url: "v".to_string(), + part_size: 8_388_608 + }) + ); + } +} diff --git a/src-tauri/src/sync/engine.rs b/src-tauri/src/sync/engine.rs index 870a67eb..d05402e9 100644 --- a/src-tauri/src/sync/engine.rs +++ b/src-tauri/src/sync/engine.rs @@ -753,6 +753,7 @@ impl SyncEngine { .client .upload_bytes_with_metadata( &presign.url, + presign.transfer.as_ref(), &payload, Some(content_type), presign.metadata.as_ref(), @@ -1324,7 +1325,12 @@ impl SyncEngine { self .client - .upload_bytes(&presign.url, &upload_data, Some(content_type)) + .upload_bytes( + &presign.url, + presign.transfer.as_ref(), + &upload_data, + Some(content_type), + ) .await?; Ok(()) @@ -1525,10 +1531,10 @@ impl SyncEngine { let batch_response = self.client.presign_upload_batch(items).await?; // Build URL map - let url_map: HashMap = batch_response + let url_map: HashMap)> = batch_response .items .into_iter() - .map(|item| (item.key, item.url)) + .map(|item| (item.key, (item.url, item.transfer))) .collect(); let total_bytes: u64 = files.iter().map(|f| f.size).sum(); @@ -1602,7 +1608,7 @@ impl SyncEngine { continue; } - let url = url.unwrap(); + let (url, transfer) = url.unwrap(); let client = client.clone(); let tracker = tracker.clone(); let resume_state = resume_state.clone(); @@ -1653,7 +1659,7 @@ impl SyncEngine { let mut last_err = String::new(); for attempt in 0..MAX_FILE_RETRIES { match client - .upload_bytes(&url, &upload_data, content_type.as_deref()) + .upload_bytes(&url, transfer.as_ref(), &upload_data, content_type.as_deref()) .await { Ok(()) => { @@ -2674,7 +2680,12 @@ impl SyncEngine { .await?; self .client - .upload_bytes(&file_presign.url, &file_payload, Some(file_content_type)) + .upload_bytes( + &file_presign.url, + file_presign.transfer.as_ref(), + &file_payload, + Some(file_content_type), + ) .await?; } @@ -4961,7 +4972,7 @@ mod tests { ?X-Amz-Signature=deadbeef"; let error = client - .upload_bytes(presigned, b"payload", None) + .upload_bytes(presigned, None, b"payload", None) .await .expect_err("a host that cannot resolve must not report a successful upload"); diff --git a/src-tauri/src/sync/types.rs b/src-tauri/src/sync/types.rs index 69e08681..1f7f2c3a 100644 --- a/src-tauri/src/sync/types.rs +++ b/src-tauri/src/sync/types.rs @@ -41,6 +41,20 @@ pub struct PresignUploadResponse { /// from older servers → client sends no metadata headers (body-GET fallback). #[serde(default)] pub metadata: Option>, + /// Set by a server that encrypts at rest: upload through it instead of + /// `url`, in parts when the file is larger than one. + #[serde(default)] + pub transfer: Option, +} + +/// A server's own upload URL, for a store it encrypts at rest. A request +/// through a proxy in front of it has a size and time limit, so a file larger +/// than `part_size` goes up in parts (`SyncClient::upload_bytes`). +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct SealedTransfer { + pub url: String, + #[serde(rename = "partSize")] + pub part_size: u64, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -126,6 +140,8 @@ pub struct PresignUploadBatchItemResponse { pub url: String, #[serde(rename = "expiresAt")] pub expires_at: String, + #[serde(default)] + pub transfer: Option, } #[derive(Debug, Clone, Serialize, Deserialize)] diff --git a/src/components/agent-tasks/batch-card.tsx b/src/components/agent-tasks/batch-card.tsx deleted file mode 100644 index 606be90e..00000000 --- a/src/components/agent-tasks/batch-card.tsx +++ /dev/null @@ -1,244 +0,0 @@ -"use client"; - -import { motion } from "motion/react"; -import { useState } from "react"; -import { useTranslation } from "react-i18next"; -import { LuRotateCcw, LuSquare, LuX } from "react-icons/lu"; -import { ProgressBar } from "@/components/agent-console-parts"; -import { useNumber } from "@/components/agent-tasks/parts"; -import { StatusDot } from "@/components/cookie-bot-shared"; -import { Button } from "@/components/ui/button"; -import { useMotionAllowed } from "@/hooks/use-motion-allowed"; -import { - type AgentBatch, - type AgentTask, - batchProgress, - dismissAgentBatch, - promptTitle, - retryAgentBatch, - statusTone, - stopAgentBatch, -} from "@/lib/agent-tasks"; -import { translateBackendError } from "@/lib/backend-errors"; -import { MOTION_SPRING } from "@/lib/motion"; -import { showErrorToast } from "@/lib/toast-utils"; -import { cn } from "@/lib/utils"; - -/** The one line a tile shows: the newest step, or what the status means. */ -function statusLine( - t: ReturnType["t"], - task: AgentTask, -): string { - if (task.status === "needs_approval") return t("agent.tasks.needsYou"); - if (task.status === "done" && task.result) return task.result; - if (task.lastStep && task.status !== "queued") return task.lastStep.text; - return t(`agent.tasks.status.${task.status}`); -} - -function ProfileTile({ - task, - name, - selected, - onSelect, -}: { - task: AgentTask; - name: string; - selected: boolean; - onSelect: () => void; -}) { - const { t } = useTranslation(); - const format = useNumber(); - const live = task.status === "running" || task.status === "launching"; - return ( - - ); -} - -/** One run on one or many profiles: totals, the batch controls and a tile per profile. */ -export function BatchCard({ - batch, - profileNames, - selectedProfileId, - onSelect, - onDismissed, -}: { - batch: AgentBatch; - profileNames: Map; - selectedProfileId: string | null; - onSelect: (profileId: string) => void; - onDismissed: (batchId: string) => void; -}) { - const { t } = useTranslation(); - const format = useNumber(); - const animate = useMotionAllowed(); - const [busy, setBusy] = useState(false); - const { tally } = batch; - const total = batch.tasks.length; - const ended = tally.done + tally.failed + tally.stopped; - const nameOf = (profileId: string) => - profileNames.get(profileId) ?? t("agent.tasks.unknownProfile"); - - const act = async (work: () => Promise) => { - setBusy(true); - try { - await work(); - } catch (error) { - showErrorToast(translateBackendError(t, error)); - } finally { - setBusy(false); - } - }; - - return ( - -
-
-

- {promptTitle(batch.prompt)} -

-

- {t("agent.tasks.progress", { done: ended, total })} - {tally.failed > 0 && ( - - {t("agent.tasks.failedCount", { n: tally.failed })} - - )} - {tally.creditsUsed > 0 && ( - - {t("agent.tasks.credits", { n: format(tally.creditsUsed) })} - - )} - {batch.routineId && {t("agent.tasks.fromRoutine")}} -

-
-
- {!batch.finished && ( - - )} - {tally.failed > 0 && batch.finished && ( - - )} - {batch.finished && ( - - )} -
-
- {total > 1 && ( - <> - -
- {batch.tasks.map((task) => ( - { - onSelect(task.profileId); - }} - /> - ))} -
- - )} - {total === 1 && ( - { - onSelect(batch.tasks[0].profileId); - }} - /> - )} - - ); -}