Skip to main content

max / synckit

29.8 KB · 719 lines History Blame Raw
1 //! Content-addressed blob upload and download.
2 //!
3 //! Blobs are encrypted client-side and stored under the SHA-256 of their
4 //! plaintext. Uploads go through a presigned URL after a size/quota check;
5 //! downloads are decrypted and re-hashed on arrival and discarded on any
6 //! mismatch, so a server that swaps or rolls back a blob cannot slip it past
7 //! the client.
8
9 use std::path::Path;
10
11 use base64::Engine;
12 use bytes::{Buf, Bytes, BytesMut};
13 use sha2::{Digest, Sha256};
14 use tokio::io::AsyncReadExt;
15 use tokio_stream::StreamExt;
16 use tracing::instrument;
17
18 use crate::{
19 crypto,
20 error::{Result, SyncKitError},
21 types::{
22 BlobConfirmRequest, BlobDownloadUrlRequest, BlobDownloadUrlResponse,
23 BlobMultipartAbortRequest, BlobMultipartCompleteRequest, BlobMultipartCompletedPart,
24 BlobMultipartPartsRequest, BlobMultipartPartsResponse, BlobMultipartStartRequest,
25 BlobMultipartStartResponse, BlobUploadUrlRequest, BlobUploadUrlResponse,
26 },
27 };
28
29 use super::SyncKitClient;
30 use super::helpers::{Idempotency, check_response};
31
32 /// Upper bound on a single decrypted blob held in memory, guarding against a
33 /// hostile server returning an absurdly large body that would OOM the client.
34 /// Generous (4 GiB) so legitimate large media still flow.
35 const MAX_BLOB_BYTES: usize = 4 * 1024 * 1024 * 1024;
36
37 /// What [`SyncKitClient::blob_upload_streaming`] did.
38 ///
39 /// The caller needs to tell the two apart: only an upload that sent bytes has
40 /// anything to confirm.
41 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
42 pub enum BlobUploadOutcome {
43 /// The parts were uploaded and assembled. Call [`SyncKitClient::blob_confirm`]
44 /// next to record the blob server-side.
45 Uploaded,
46 /// The server already held this content address, so no bytes were sent and
47 /// there is nothing to confirm. Server-asserted, not verified, see
48 /// [`BlobUploadUrlResponse::already_exists`] for what a wrong `true` costs.
49 AlreadyPresent,
50 }
51
52 impl SyncKitClient {
53 /// Request a presigned upload URL for a blob.
54 /// Returns (upload_url, already_exists). If `already_exists` is true,
55 /// the blob is already on the server and no upload is needed.
56 ///
57 /// `size_bytes` is the **plaintext** length, the size the caller sees on
58 /// disk. What [`Self::blob_upload`] actually PUTs is the sealed ciphertext,
59 /// which is longer by the v3 framing, and the server signs the declared
60 /// size into the presigned URL as `Content-Length` (a SignedHeader), so
61 /// declaring the plaintext length makes every PUT fail the signature. This
62 /// method converts, keeping the wire-format arithmetic inside the SDK
63 /// instead of asking three consumer apps to know it.
64 ///
65 /// `already_exists` is the server's claim, not a verified fact. See
66 /// [`BlobUploadUrlResponse::already_exists`] for what a false `true` costs
67 /// (a withheld blob, never a substituted one).
68 #[instrument(skip(self))]
69 pub async fn blob_upload_url(
70 &self,
71 hash: &str,
72 size_bytes: i64,
73 ) -> Result<BlobUploadUrlResponse> {
74 let plaintext_len = usize::try_from(size_bytes)
75 .map_err(|_| SyncKitError::InvalidArgument("size_bytes must be non-negative".into()))?;
76 let token = self.require_token()?;
77
78 let body = Bytes::from(serde_json::to_vec(&BlobUploadUrlRequest {
79 hash: hash.to_string(),
80 size_bytes: crypto::blob_encrypted_len(plaintext_len) as i64,
81 })?);
82
83 self.retry_request_json(Idempotency::ReadOnly, || {
84 let req = self
85 .http
86 .post(&self.endpoints.blobs_upload)
87 .bearer_auth(&token)
88 .header("content-type", "application/json")
89 .body(body.clone());
90 async move { check_response(req.send().await?).await }
91 })
92 .await
93 }
94
95 /// Upload blob data directly to S3 via a presigned PUT URL.
96 ///
97 /// Encrypts the data with the master key before uploading, binding the
98 /// content `hash` as AEAD associated data so the ciphertext is tied to its
99 /// content address. The plaintext is never sent to the server, preserving
100 /// the E2E encryption guarantee.
101 ///
102 /// This is the in-memory path: it buffers the whole ciphertext, so the input
103 /// is capped at [`MAX_BLOB_BYTES`] rather than risking an unbounded
104 /// allocation. For anything file-backed, prefer
105 /// [`Self::blob_upload_streaming`], which holds one part at a time and is
106 /// the only path the server accepts above its one-shot PUT ceiling.
107 #[instrument(skip(self, presigned_url, data))]
108 pub async fn blob_upload(&self, hash: &str, presigned_url: &str, data: Vec<u8>) -> Result<()> {
109 if data.len() > MAX_BLOB_BYTES {
110 return Err(SyncKitError::InvalidArgument(format!(
111 "blob is {} bytes, over the {MAX_BLOB_BYTES}-byte in-memory cap; use blob_upload_streaming for a file this size",
112 data.len()
113 )));
114 }
115 let master_key = self.require_master_key()?;
116 // v3 chunked format: each chunk sealed and bound to (hash, index, count),
117 // so the reader verifies and decrypts one chunk at a time.
118 let encrypted = Bytes::from(crypto::encrypt_blob_chunked(&data, &master_key, hash)?);
119 // Release the plaintext before the (potentially long, multi-GB) network
120 // send so the sustained peak during transfer is ~1x the blob, not ~2x.
121 // The per-attempt `encrypted.clone()` below is a cheap `Bytes` refcount
122 // bump, not a copy. Never materializing the full ciphertext at all is
123 // what `blob_upload_streaming` does, for callers holding a file.
124 drop(data);
125
126 self.retry_request(
127 Idempotency::IdempotentWrite {
128 on: "content hash (S3 key is content-addressed)",
129 },
130 || {
131 let req = self
132 .http_stream
133 .put(presigned_url)
134 .header("content-type", "application/octet-stream")
135 .body(encrypted.clone());
136 async move { check_response(req.send().await?).await }
137 },
138 )
139 .await?;
140
141 Ok(())
142 }
143
144 /// Upload a file-backed blob through the multipart session, never holding
145 /// more than one part in memory.
146 ///
147 /// This is the bounded-memory counterpart to [`Self::blob_upload`], and the
148 /// only path for blobs above the server's one-shot PUT ceiling. It works at
149 /// any size (a small blob is a single part), so it is the better default
150 /// whenever the blob is already a file on disk.
151 ///
152 /// The v3 chunk geometry is what makes it possible. A presigned PUT needs a
153 /// signed `Content-Length` up front, and `blob_encrypted_len` gives the
154 /// exact ciphertext length from the plaintext length alone, so the server
155 /// signs every part boundary before a single byte has been read, and this
156 /// method fills those boundaries by sealing 1 MiB chunks as it streams the
157 /// file. Peak memory is one part plus one chunk, not the whole blob.
158 ///
159 /// `hash` must be the SHA-256 of the file's plaintext, as elsewhere. It is
160 /// recomputed while streaming and the upload is abandoned on mismatch: the
161 /// caller hashed the file in an earlier pass, so a file edited in between
162 /// would otherwise be stored under an address that does not describe it.
163 ///
164 /// Returns [`BlobUploadOutcome`]: on `Uploaded` the caller calls
165 /// [`Self::blob_confirm`] next, exactly as with the one-shot path (this
166 /// method replaces the transport only); on `AlreadyPresent` the server
167 /// already held the content, nothing was read or sent, and there is nothing
168 /// to confirm. The dedup check happens before any file read or sealing, so
169 /// a re-sync of content the server already has costs one round trip.
170 #[instrument(skip(self, path), fields(path = %path.display()))]
171 pub async fn blob_upload_streaming(
172 &self,
173 hash: &str,
174 path: &Path,
175 ) -> Result<BlobUploadOutcome> {
176 let master_key = self.require_master_key()?;
177
178 let meta = tokio::fs::metadata(path).await.map_err(|e| {
179 SyncKitError::Internal(format!("stat blob file {}: {e}", path.display()))
180 })?;
181 let plaintext_len = usize::try_from(meta.len())
182 .map_err(|_| SyncKitError::InvalidArgument("blob file is larger than usize".into()))?;
183 if plaintext_len > MAX_BLOB_BYTES {
184 return Err(SyncKitError::InvalidArgument(format!(
185 "blob is {plaintext_len} bytes, over the {MAX_BLOB_BYTES}-byte client cap"
186 )));
187 }
188 // The session is sized in ciphertext, which is knowable from the
189 // plaintext length alone, that is the whole trick.
190 let size_bytes = i64::try_from(crypto::blob_encrypted_len(plaintext_len))
191 .map_err(|_| SyncKitError::InvalidArgument("blob ciphertext exceeds i64".into()))?;
192
193 let token = self.require_token()?;
194 let start_body = Bytes::from(serde_json::to_vec(&BlobMultipartStartRequest {
195 hash: hash.to_string(),
196 size_bytes,
197 })?);
198 let start: BlobMultipartStartResponse = self
199 .retry_request_json(Idempotency::ReadOnly, || {
200 let req = self
201 .http
202 .post(&self.endpoints.blobs_multipart_start)
203 .bearer_auth(&token)
204 .header("content-type", "application/json")
205 .body(start_body.clone());
206 async move { check_response(req.send().await?).await }
207 })
208 .await?;
209
210 if start.already_exists {
211 return Ok(BlobUploadOutcome::AlreadyPresent);
212 }
213
214 // Every failure past this point leaves uploaded parts that S3 bills for
215 // until the session is aborted, so the abort lives in exactly one place:
216 // here, around the whole transfer. The server-side orphan reaper is the
217 // backstop for a client that dies outright.
218 let transfer = self
219 .stream_blob_parts(hash, &start, path, plaintext_len, size_bytes, &master_key)
220 .await;
221 let parts = match transfer {
222 Ok(parts) => parts,
223 Err(e) => {
224 if let Err(abort_err) = self.blob_multipart_abort(hash, &start.upload_id).await {
225 tracing::warn!(
226 error = %abort_err,
227 "failed to abort multipart blob session; the server reaper will release it"
228 );
229 }
230 return Err(e);
231 }
232 };
233
234 let complete_body = Bytes::from(serde_json::to_vec(&BlobMultipartCompleteRequest {
235 hash: hash.to_string(),
236 upload_id: start.upload_id.clone(),
237 parts,
238 })?);
239 self.retry_request(
240 Idempotency::IdempotentWrite {
241 on: "upload_id (completing twice assembles the same object)",
242 },
243 || {
244 let req = self
245 .http
246 .post(&self.endpoints.blobs_multipart_complete)
247 .bearer_auth(&token)
248 .header("content-type", "application/json")
249 .body(complete_body.clone());
250 async move { check_response(req.send().await?).await }
251 },
252 )
253 .await?;
254
255 Ok(BlobUploadOutcome::Uploaded)
256 }
257
258 /// Seal the file chunk by chunk, cutting the ciphertext at the server's
259 /// signed part boundaries and PUTting each part. Returns the completed
260 /// `(part_number, etag)` pairs.
261 ///
262 /// Peak memory is one part plus one chunk: the staging buffer never holds
263 /// more than a part's worth, because a full part is drained and sent the
264 /// moment it is complete.
265 async fn stream_blob_parts(
266 &self,
267 hash: &str,
268 start: &BlobMultipartStartResponse,
269 path: &Path,
270 plaintext_len: usize,
271 size_bytes: i64,
272 master_key: &[u8; 32],
273 ) -> Result<Vec<BlobMultipartCompletedPart>> {
274 let part_size = usize::try_from(start.part_size).map_err(|_| {
275 SyncKitError::Internal(format!(
276 "server part_size {} exceeds usize",
277 start.part_size
278 ))
279 })?;
280 if part_size == 0 || start.part_count == 0 {
281 return Err(SyncKitError::Internal(
282 "server returned an empty multipart plan".into(),
283 ));
284 }
285
286 // The part geometry is server-supplied, so bound it before it drives any
287 // allocation. Without this a hostile part_count aborts the client on an
288 // over-large `Vec`/`BytesMut` reservation, and a part_count/part_size that
289 // does not match the blob defeats the one-part-plus-one-chunk memory bound.
290 // The plan must describe exactly the ciphertext we are about to upload
291 // (the server derives it from this same `size_bytes`), so recompute and
292 // compare rather than trust it.
293 const MAX_PART_SIZE: usize = 1 << 30; // 1 GiB; real plan parts are well under this
294 const MAX_PART_COUNT: u32 = 10_000; // S3's hard multipart-part ceiling
295 if part_size > MAX_PART_SIZE {
296 return Err(SyncKitError::Internal(format!(
297 "server part_size {part_size} exceeds the {MAX_PART_SIZE}-byte ceiling"
298 )));
299 }
300 if start.part_count > MAX_PART_COUNT {
301 return Err(SyncKitError::Internal(format!(
302 "server part_count {} exceeds the {MAX_PART_COUNT}-part ceiling",
303 start.part_count
304 )));
305 }
306 let total = usize::try_from(size_bytes).map_err(|_| {
307 SyncKitError::Internal(format!("blob size {size_bytes} is not a valid length"))
308 })?;
309 let expected_parts = total.div_ceil(part_size);
310 if start.part_count as usize != expected_parts {
311 return Err(SyncKitError::Internal(format!(
312 "server multipart plan (part_count {}, part_size {part_size}) does not match the \
313 {total}-byte blob (expected {expected_parts} parts)",
314 start.part_count
315 )));
316 }
317
318 let mut file = tokio::fs::File::open(path).await.map_err(|e| {
319 SyncKitError::Internal(format!("open blob file {}: {e}", path.display()))
320 })?;
321
322 let chunk_count = crypto::blob_chunk_count_for(plaintext_len);
323 let mut hasher = Sha256::new();
324 // Plaintext scratch: zeroized on drop so a decrypted chunk does not linger.
325 let mut plain = zeroize::Zeroizing::new(vec![0u8; crypto::BLOB_CHUNK_SIZE]);
326 let mut staged = BytesMut::with_capacity(part_size + crypto::BLOB_CHUNK_SIZE);
327 staged.extend_from_slice(&crypto::blob_header_bytes(plaintext_len));
328
329 let mut completed = Vec::with_capacity(start.part_count as usize);
330 let mut next_part = 1u32;
331
332 for index in 0..chunk_count {
333 let offset = index as usize * crypto::BLOB_CHUNK_SIZE;
334 let want = (plaintext_len - offset).min(crypto::BLOB_CHUNK_SIZE);
335 file.read_exact(&mut plain[..want]).await.map_err(|e| {
336 SyncKitError::Internal(format!(
337 "read blob file {} at offset {offset}: {e}",
338 path.display()
339 ))
340 })?;
341 hasher.update(&plain[..want]);
342 staged.extend_from_slice(&crypto::seal_blob_chunk(
343 &plain[..want],
344 master_key,
345 hash,
346 index,
347 chunk_count,
348 )?);
349
350 // Every part but the last is exactly `part_size`; the remainder is
351 // the final part, so it is never sent from inside this loop.
352 while staged.len() >= part_size && next_part < start.part_count {
353 let body = staged.split_to(part_size).freeze();
354 let etag = self
355 .upload_blob_part(hash, start, size_bytes, next_part, body)
356 .await?;
357 completed.push(BlobMultipartCompletedPart {
358 part_number: next_part as i32,
359 etag,
360 });
361 next_part += 1;
362 }
363 }
364
365 // Verify the content address before the object can be assembled. A
366 // mismatch here aborts the session, so a file that changed after the
367 // caller hashed it is never stored under the stale address.
368 let actual = hex::encode(hasher.finalize());
369 if actual != hash {
370 return Err(SyncKitError::IntegrityFailed {
371 expected: hash.to_string(),
372 actual,
373 });
374 }
375
376 if next_part != start.part_count {
377 return Err(SyncKitError::Internal(format!(
378 "sealed stream produced {} parts, server planned {}",
379 next_part, start.part_count
380 )));
381 }
382 let body = staged.split().freeze();
383 let etag = self
384 .upload_blob_part(hash, start, size_bytes, next_part, body)
385 .await?;
386 completed.push(BlobMultipartCompletedPart {
387 part_number: next_part as i32,
388 etag,
389 });
390
391 Ok(completed)
392 }
393
394 /// PUT one part to its presigned URL and return the ETag S3 assigned it.
395 ///
396 /// The URL is requested for this part alone, and carries the part's SHA-256
397 /// bound as `x-amz-checksum-sha256`, so S3 rehashes what it receives and
398 /// rejects a corrupted part at write time instead of letting it sit until a
399 /// download fails to decrypt. That is the reason parts are not requested in
400 /// windows: a digest only exists once the part has been sealed, and only one
401 /// part is ever in memory. The cost is one small round trip per part.
402 async fn upload_blob_part(
403 &self,
404 hash: &str,
405 start: &BlobMultipartStartResponse,
406 size_bytes: i64,
407 part_number: u32,
408 body: Bytes,
409 ) -> Result<String> {
410 let checksum = base64::engine::general_purpose::STANDARD.encode(Sha256::digest(&body));
411
412 let token = self.require_token()?;
413 let req_body = Bytes::from(serde_json::to_vec(&BlobMultipartPartsRequest {
414 hash: hash.to_string(),
415 upload_id: start.upload_id.clone(),
416 size_bytes,
417 first_part: part_number,
418 count: 1,
419 checksums: vec![checksum.clone()],
420 })?);
421 let window: BlobMultipartPartsResponse = self
422 .retry_request_json(Idempotency::ReadOnly, || {
423 let req = self
424 .http
425 .post(&self.endpoints.blobs_multipart_parts)
426 .bearer_auth(&token)
427 .header("content-type", "application/json")
428 .body(req_body.clone());
429 async move { check_response(req.send().await?).await }
430 })
431 .await?;
432
433 let part = window.parts.into_iter().next().ok_or_else(|| {
434 SyncKitError::Internal(format!("server returned no URL for part {part_number}"))
435 })?;
436 // The two sides derive the plan independently, so a disagreement means
437 // one of them is wrong about the geometry. Sending anyway would fail the
438 // signed Content-Length at S3 with a far less legible error.
439 if part.part_number != part_number as i32 || part.content_length != body.len() as u64 {
440 return Err(SyncKitError::Internal(format!(
441 "part geometry mismatch: server signed part {} for {} bytes, client has part {part_number} of {} bytes",
442 part.part_number,
443 part.content_length,
444 body.len()
445 )));
446 }
447
448 let resp = self
449 .retry_request(
450 Idempotency::IdempotentWrite {
451 on: "(upload_id, part_number), re-PUTting a part replaces it",
452 },
453 || {
454 let req = self
455 .http_stream
456 .put(&part.url)
457 .header("content-type", "application/octet-stream")
458 // Mandatory, not optional: the checksum is a signed
459 // header, so omitting it fails the signature.
460 .header("x-amz-checksum-sha256", &checksum)
461 .body(body.clone());
462 async move { check_response(req.send().await?).await }
463 },
464 )
465 .await?;
466
467 // S3 identifies each part by the ETag it returns; complete cannot be
468 // assembled without it.
469 resp.headers()
470 .get(reqwest::header::ETAG)
471 .and_then(|v| v.to_str().ok())
472 .map(std::string::ToString::to_string)
473 .ok_or_else(|| {
474 SyncKitError::Internal(format!("part {part_number} upload returned no ETag"))
475 })
476 }
477
478 /// Release an abandoned multipart session and the parts it holds.
479 async fn blob_multipart_abort(&self, hash: &str, upload_id: &str) -> Result<()> {
480 let token = self.require_token()?;
481 let body = Bytes::from(serde_json::to_vec(&BlobMultipartAbortRequest {
482 hash: hash.to_string(),
483 upload_id: upload_id.to_string(),
484 })?);
485 self.retry_request(
486 Idempotency::IdempotentWrite {
487 on: "upload_id (aborting twice is a no-op)",
488 },
489 || {
490 let req = self
491 .http
492 .post(&self.endpoints.blobs_multipart_abort)
493 .bearer_auth(&token)
494 .header("content-type", "application/json")
495 .body(body.clone());
496 async move { check_response(req.send().await?).await }
497 },
498 )
499 .await?;
500 Ok(())
501 }
502
503 /// Confirm that a blob upload completed successfully.
504 /// The server verifies the object exists in S3 and records it.
505 #[instrument(skip(self))]
506 pub async fn blob_confirm(&self, hash: &str, size_bytes: i64) -> Result<()> {
507 if size_bytes < 0 {
508 return Err(SyncKitError::InvalidArgument(
509 "size_bytes must be non-negative".into(),
510 ));
511 }
512 let token = self.require_token()?;
513
514 let body = Bytes::from(serde_json::to_vec(&BlobConfirmRequest {
515 hash: hash.to_string(),
516 size_bytes,
517 })?);
518
519 self.retry_request(Idempotency::IdempotentWrite { on: "blob hash" }, || {
520 let req = self
521 .http
522 .post(&self.endpoints.blobs_confirm)
523 .bearer_auth(&token)
524 .header("content-type", "application/json")
525 .body(body.clone());
526 async move { check_response(req.send().await?).await }
527 })
528 .await?;
529 Ok(())
530 }
531
532 /// Get a presigned download URL for a blob by hash.
533 #[instrument(skip(self))]
534 pub async fn blob_download_url(&self, hash: &str) -> Result<String> {
535 let token = self.require_token()?;
536
537 let body = Bytes::from(serde_json::to_vec(&BlobDownloadUrlRequest {
538 hash: hash.to_string(),
539 })?);
540
541 let download: BlobDownloadUrlResponse = self
542 .retry_request_json(Idempotency::ReadOnly, || {
543 let req = self
544 .http
545 .post(&self.endpoints.blobs_download)
546 .bearer_auth(&token)
547 .header("content-type", "application/json")
548 .body(body.clone());
549 async move { check_response(req.send().await?).await }
550 })
551 .await?;
552 Ok(download.download_url)
553 }
554
555 /// Download blob data from S3 via a presigned GET URL and verify it.
556 ///
557 /// Decrypts with the master key (binding the content `expected_hash` as AEAD
558 /// associated data), then **re-verifies** that the plaintext hashes to
559 /// `expected_hash`. The AEAD tag alone proves the bytes were not forged, but
560 /// not that the untrusted server served the bytes for *this* address; the
561 /// re-hash closes the substitution/rollback gap. Returns
562 /// [`SyncKitError::IntegrityFailed`] on mismatch.
563 ///
564 /// The decrypted blob is returned in memory (the caller writes it to disk);
565 /// the size is capped at [`MAX_BLOB_BYTES`]. A v3 chunked blob is decrypted
566 /// chunk-by-chunk off the HTTP body, so peak memory is the plaintext plus one
567 /// in-flight chunk rather than ciphertext *and* plaintext at once. Legacy
568 /// `sk2:`/untagged blobs fall back to a whole-buffer decrypt (no flag-day).
569 #[instrument(skip(self, presigned_url))]
570 pub async fn blob_download(&self, expected_hash: &str, presigned_url: &str) -> Result<Vec<u8>> {
571 let resp = self
572 .retry_request(Idempotency::ReadOnly, || {
573 let req = self.http_stream.get(presigned_url);
574 async move { check_response(req.send().await?).await }
575 })
576 .await?;
577
578 let master_key = self.require_master_key()?;
579 let mut stream = resp.bytes_stream();
580 let mut buf = BytesMut::new();
581 let mut header: Option<crypto::BlobHeader> = None;
582 let mut format_known = false;
583 let mut chunked = false;
584 let mut next_chunk: u32 = 0;
585 let mut plaintext: Vec<u8> = Vec::new();
586 let mut total_in: usize = 0;
587
588 while let Some(part) = stream.next().await {
589 let part = part.map_err(SyncKitError::Http)?;
590 total_in = total_in.saturating_add(part.len());
591 if total_in > MAX_BLOB_BYTES {
592 return Err(SyncKitError::Internal(format!(
593 "blob exceeds {MAX_BLOB_BYTES}-byte limit",
594 )));
595 }
596 buf.extend_from_slice(&part);
597
598 // Pick the format once the 4-byte tag is in hand.
599 if !format_known && buf.len() >= 4 {
600 chunked = crypto::is_chunked_blob(&buf);
601 format_known = true;
602 }
603 if !chunked {
604 continue; // legacy: accumulate, decrypt whole-buffer at the end
605 }
606 // Parse the v3 header once available, consuming tag + header bytes.
607 if header.is_none() {
608 if buf.len() < 4 + 13 {
609 continue;
610 }
611 let (h, consumed) = crypto::parse_blob_header(&buf)?;
612 buf.advance(consumed);
613 header = Some(h);
614 }
615 // Peel and decrypt every complete chunk currently buffered.
616 let h = header.unwrap();
617 while next_chunk < h.chunk_count {
618 let len = h.sealed_chunk_len(next_chunk);
619 if buf.len() < len {
620 break;
621 }
622 let sealed = buf.split_to(len);
623 let chunk = crypto::decrypt_blob_chunk(
624 &sealed,
625 &master_key,
626 expected_hash,
627 next_chunk,
628 h.chunk_count,
629 )?;
630 plaintext.extend_from_slice(&chunk);
631 next_chunk += 1;
632 }
633 }
634
635 if chunked {
636 let h = header
637 .ok_or_else(|| SyncKitError::Crypto("v3 blob ended before its header".into()))?;
638 if next_chunk != h.chunk_count || buf.has_remaining() {
639 return Err(SyncKitError::Crypto(
640 "v3 blob ended mid-chunk or had trailing bytes".into(),
641 ));
642 }
643 } else {
644 // Legacy (sk2: / untagged): decrypt the accumulated body in one shot.
645 plaintext = crypto::decrypt_bytes_aad(
646 &buf,
647 &master_key,
648 &crypto::AeadContext::blob(expected_hash),
649 )?;
650 }
651
652 let actual = hex::encode(Sha256::digest(&plaintext));
653 if actual != expected_hash {
654 return Err(SyncKitError::IntegrityFailed {
655 expected: expected_hash.to_string(),
656 actual,
657 });
658 }
659 Ok(plaintext)
660 }
661 }
662
663 #[cfg(test)]
664 mod tests {
665 use crate::types::*;
666
667 #[test]
668 fn blob_upload_url_response_deserialization() {
669 let json = r#"{"upload_url": "https://s3.example.com/upload", "already_exists": false}"#;
670 let resp: BlobUploadUrlResponse = serde_json::from_str(json).unwrap();
671 assert_eq!(resp.upload_url, "https://s3.example.com/upload");
672 assert!(!resp.already_exists);
673
674 let json = r#"{"upload_url": "", "already_exists": true}"#;
675 let resp: BlobUploadUrlResponse = serde_json::from_str(json).unwrap();
676 assert!(resp.already_exists);
677 }
678
679 #[test]
680 fn blob_upload_url_request_serialization() {
681 let req = BlobUploadUrlRequest {
682 hash: "sha256-abc123".to_string(),
683 size_bytes: 1024,
684 };
685
686 let json = serde_json::to_string(&req).unwrap();
687 let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
688 assert_eq!(parsed["hash"], "sha256-abc123");
689 assert_eq!(parsed["size_bytes"], 1024);
690 }
691
692 #[test]
693 fn blob_content_hash_format_matches_consumer() {
694 use sha2::{Digest, Sha256};
695 // The integrity check in blob_download compares against this exact form:
696 // lowercase hex of SHA-256, the same string consumers store as the blob
697 // hash. If this drifts, every verified download would falsely reject.
698 let h = hex::encode(Sha256::digest(b"hello blob"));
699 assert_eq!(h.len(), 64);
700 assert!(
701 h.chars()
702 .all(|c| c.is_ascii_hexdigit() && !c.is_ascii_uppercase())
703 );
704 }
705
706 #[test]
707 fn blob_confirm_request_serialization() {
708 let req = BlobConfirmRequest {
709 hash: "sha256-def456".to_string(),
710 size_bytes: 2048,
711 };
712
713 let json = serde_json::to_string(&req).unwrap();
714 let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
715 assert_eq!(parsed["hash"], "sha256-def456");
716 assert_eq!(parsed["size_bytes"], 2048);
717 }
718 }
719