Skip to main content

max / makenotwork

Fix the blob presign size declaration, bind part checksums Two things about the presigned-PUT contract, found while implementing the deferred checksum item. The bug: presign_upload signs max_bytes as Content-Length, and it really is a SignedHeader (a presigned URL comes back with X-Amz-SignedHeaders=content-length;content-type;host, now pinned by a test). Both consumers declared the plaintext size while blob_upload PUT the sealed ciphertext, which is always longer by the v3 framing — so every one-shot blob upload would fail SigV4 against a real backend. Nothing caught it: the server tests use an in-memory backend that does not sign, and the client tests use wiremock, which does not verify. blob_upload_url now converts to blob_encrypted_len, keeping the wire-format arithmetic in the SDK rather than in three consumer apps, and a test pins the declared size to the actual PUT body length. The checksum: the deferred note claimed binding x-amz-checksum-sha256 would make S3 enforce the content address. It cannot. The stored object is E2E ciphertext with random per-chunk nonces while the address is the SHA-256 of the plaintext, so the server cannot derive the expected digest and any checksum it binds comes from the client — the party whose honesty was in question. What binds bytes to the address is the AEAD, which already seals each chunk under (hash, index, chunk_count). The note is rewritten to say so. Bound anyway, for transport integrity: S3 rehashes each part and rejects a corrupted one before the bytes are durable. Multipart only, and the client now requests one part URL at a time, since a digest exists only once a part is sealed and only one part is ever in memory. The one-shot path keeps no checksum: binding one there would mean sealing before requesting the URL, wasting a full read+seal on every dedup hit. File-backed callers should move to blob_upload_streaming, whose start does dedup before any work. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-21 15:31 UTC
Commit: 436a4eb1003bcd29ad43a2941b263ef7d5019414
Parent: 99d418b
11 files changed, +443 insertions, -80 deletions
M server/src/routes/api/internal/uploads.rs +5
@@ -264,6 +264,11 @@ pub(super) async fn multipart_parts(
264 264 part_number as i32,
265 265 Some(expires_in),
266 266 Some(content_length as i64),
267 + // No checksum: the CLI streams plain file bytes and asks for
268 + // part URLs ahead of reading them, so it has no digest to bind
269 + // yet. The synckit blob path, which seals a part before asking
270 + // for its URL, does bind one.
271 + None,
267 272 )
268 273 .await?;
269 274 parts.push(MultipartPartUrl {
M server/src/routes/synckit/blobs.rs +43 -11
@@ -273,6 +273,20 @@ pub(super) async fn blob_multipart_parts(
273 273 )));
274 274 }
275 275
276 + // Checksums are positional, so a short or long list would silently bind the
277 + // wrong digest to a part — reject it rather than guess the alignment.
278 + if let Some(checksums) = &req.checksums {
279 + if checksums.len() != req.count as usize {
280 + return Err(AppError::BadRequest(format!(
281 + "checksums must have exactly {} entries, one per requested part",
282 + req.count
283 + )));
284 + }
285 + for c in checksums {
286 + validation::validate_sha256_base64(c)?;
287 + }
288 + }
289 +
276 290 let s3_key = crate::storage::S3Client::generate_synckit_blob_key(
277 291 sync_user.app_id, sync_user.user_id, &req.hash,
278 292 );
@@ -282,6 +296,11 @@ pub(super) async fn blob_multipart_parts(
282 296 let mut parts = Vec::with_capacity((last - req.first_part + 1) as usize);
283 297 for part_number in req.first_part..=last {
284 298 let content_length = plan.part_len(part_number);
299 + let checksum = req
300 + .checksums
301 + .as_ref()
302 + .and_then(|c| c.get((part_number - req.first_part) as usize))
303 + .map(String::as_str);
285 304 let url = synckit_s3
286 305 .presign_upload_part(
287 306 &s3_key,
@@ -289,6 +308,7 @@ pub(super) async fn blob_multipart_parts(
289 308 part_number as i32,
290 309 Some(expires_in),
291 310 Some(content_length as i64),
311 + checksum,
292 312 )
293 313 .await
294 314 .context("presign upload part for sync blob")?;
@@ -394,17 +414,29 @@ pub(super) async fn blob_multipart_abort(
394 414 /// Idempotent: returns success without creating a duplicate.
395 415 ///
396 416 /// Content-addressing trust model (ultra-fuzz Run 4 Storage NOTE, decision
397 - /// 2026-06-23): the blob `hash` is treated as a content-address LABEL — confirm
398 - /// reads the authoritative `object_size` from S3 but does not re-hash the bytes
399 - /// to prove they match `hash`. The blast radius is per-user only: the key is
400 - /// `{app_id}/{user_id}/{hash}` and storage is `UNIQUE(app_id, user_id, hash)`,
401 - /// so a client that stores mismatched bytes can poison only its OWN dedup
402 - /// namespace — no cross-user effect, no data exposure. The correct A+ fix binds
403 - /// `x-amz-checksum-sha256` into the presigned PUT so S3 itself rejects a
404 - /// mismatched upload at write time (zero server-side download); that requires the
405 - /// `synckit-client` SDK to send the checksum header, so it is deferred to the
406 - /// SDK's own remediation rather than paying a full object download + re-hash on
407 - /// every confirm here.
417 + /// 2026-06-23; revised 2026-07-21): the blob `hash` is treated as a
418 + /// content-address LABEL — confirm reads the authoritative `object_size` from S3
419 + /// but does not re-hash the bytes to prove they match `hash`. The blast radius
420 + /// is per-user only: the key is `{app_id}/{user_id}/{hash}` and storage is
421 + /// `UNIQUE(app_id, user_id, hash)`, so a client that stores mismatched bytes can
422 + /// poison only its OWN dedup namespace — no cross-user effect, no data exposure.
423 + ///
424 + /// This note used to say the A+ fix was binding `x-amz-checksum-sha256` into the
425 + /// presigned PUT so S3 rejects a mismatched upload at write time. That reasoning
426 + /// does not hold for these blobs, and the correction is worth keeping: the stored
427 + /// object is E2E *ciphertext* sealed with random per-chunk nonces, while `hash`
428 + /// is the SHA-256 of the *plaintext*. The server never sees plaintext, so it
429 + /// cannot derive the expected ciphertext digest at presign time — any checksum it
430 + /// binds has to come from the client, i.e. the party whose honesty was in
431 + /// question. Checksum binding (which the multipart path now does per part) buys
432 + /// transport integrity, not content-address enforcement.
433 + ///
434 + /// What actually binds the bytes to the address is the AEAD: each chunk is sealed
435 + /// with `(hash, chunk_index, chunk_count)` as associated data, so ciphertext that
436 + /// opens under `hash` is cryptographically tied to it, and the client re-hashes
437 + /// the plaintext after decrypting. A client storing mismatched bytes breaks only
438 + /// its own blob. Server-side re-hashing would cost a full object download per
439 + /// confirm to defend a client against itself, which is why it is not done.
408 440 #[utoipa::path(post, path = "/api/v1/sync/blobs/confirm", tag = "SyncKit",
409 441 request_body = BlobConfirmRequest,
410 442 responses((status = 204, description = "Upload confirmed")),
M server/src/routes/synckit/mod.rs +8
@@ -402,6 +402,14 @@ pub(crate) struct BlobMultipartPartsRequest {
402 402 pub size_bytes: i64,
403 403 pub first_part: u32,
404 404 pub count: u32,
405 + /// SHA-256 of each requested part's bytes, base64 of the raw digest,
406 + /// positionally aligned with `first_part..first_part + count`. Bound into
407 + /// the presigned URL so S3 rehashes the part and rejects a mismatch at write
408 + /// time. Optional for now, since a client can only supply a checksum for a
409 + /// part it has already built — which in practice means asking for one part
410 + /// at a time. When present the length must equal `count`.
411 + #[serde(default)]
412 + pub checksums: Option<Vec<String>>,
405 413 }
406 414
407 415 #[derive(Serialize, utoipa::ToSchema)]
M server/src/storage.rs +9 -5
@@ -535,7 +535,11 @@ pub trait StorageBackend: Send + Sync {
535 535 /// `max_bytes`, when set, is signed as `Content-Length`, the same
536 536 /// defense-in-depth as [`Self::presign_upload`] — the authoritative size
537 537 /// check still happens at confirm time.
538 - async fn presign_upload_part(&self, s3_key: &S3Key, upload_id: &str, part_number: i32, expiry_secs: Option<u64>, max_bytes: Option<i64>) -> Result<String>;
538 + ///
539 + /// `checksum_sha256` (base64 of the raw digest), when set, is signed as
540 + /// `x-amz-checksum-sha256` and IS enforced: S3 rehashes the part and
541 + /// rejects a mismatch before the bytes are durable.
542 + async fn presign_upload_part(&self, s3_key: &S3Key, upload_id: &str, part_number: i32, expiry_secs: Option<u64>, max_bytes: Option<i64>, checksum_sha256: Option<&str>) -> Result<String>;
539 543 /// Complete a multipart upload from the collected `(part_number, etag)`
540 544 /// pairs. Parts may be passed in any order; the backend sorts them.
541 545 async fn complete_multipart_upload(&self, s3_key: &S3Key, upload_id: &str, parts: &[(i32, String)]) -> Result<()>;
@@ -982,9 +986,9 @@ impl S3Client {
982 986
983 987 /// Presign one `UploadPart` request. See the
984 988 /// [`StorageBackend::presign_upload_part`] trait method.
985 - pub async fn presign_upload_part(&self, s3_key: &S3Key, upload_id: &str, part_number: i32, expiry_secs: Option<u64>, max_bytes: Option<i64>) -> Result<String> {
989 + pub async fn presign_upload_part(&self, s3_key: &S3Key, upload_id: &str, part_number: i32, expiry_secs: Option<u64>, max_bytes: Option<i64>, checksum_sha256: Option<&str>) -> Result<String> {
986 990 self.inner
987 - .presign_upload_part(s3_key.as_str(), upload_id, part_number, expiry_secs.unwrap_or(PRESIGN_EXPIRY_SECS), max_bytes)
991 + .presign_upload_part(s3_key.as_str(), upload_id, part_number, expiry_secs.unwrap_or(PRESIGN_EXPIRY_SECS), max_bytes, checksum_sha256)
988 992 .await
989 993 .map_err(AppError::Storage)
990 994 }
@@ -1269,8 +1273,8 @@ impl StorageBackend for S3Client {
1269 1273 self.create_multipart_upload(s3_key, content_type).await
1270 1274 }
1271 1275
1272 - async fn presign_upload_part(&self, s3_key: &S3Key, upload_id: &str, part_number: i32, expiry_secs: Option<u64>, max_bytes: Option<i64>) -> Result<String> {
1273 - self.presign_upload_part(s3_key, upload_id, part_number, expiry_secs, max_bytes).await
1276 + async fn presign_upload_part(&self, s3_key: &S3Key, upload_id: &str, part_number: i32, expiry_secs: Option<u64>, max_bytes: Option<i64>, checksum_sha256: Option<&str>) -> Result<String> {
1277 + self.presign_upload_part(s3_key, upload_id, part_number, expiry_secs, max_bytes, checksum_sha256).await
1274 1278 }
1275 1279
1276 1280 async fn complete_multipart_upload(&self, s3_key: &S3Key, upload_id: &str, parts: &[(i32, String)]) -> Result<()> {
M server/src/validation/mod.rs +20
@@ -248,6 +248,26 @@ pub fn validate_sync_blob_hash(hash: &str) -> Result<(), AppError> {
248 248 Ok(())
249 249 }
250 250
251 + /// Validate a base64-encoded SHA-256 digest, the form S3 wants for
252 + /// `x-amz-checksum-sha256`.
253 + ///
254 + /// Bound into a presigned URL, so a malformed value would produce a URL that
255 + /// fails at S3 with an opaque error long after the client committed to it.
256 + /// Reject it here, where the message can say what was wrong.
257 + pub fn validate_sha256_base64(value: &str) -> Result<(), AppError> {
258 + use base64::Engine;
259 + let decoded = base64::engine::general_purpose::STANDARD
260 + .decode(value)
261 + .map_err(|_| AppError::validation("Checksum must be base64".to_string()))?;
262 + if decoded.len() != 32 {
263 + return Err(AppError::validation(format!(
264 + "Checksum must decode to 32 bytes (SHA-256), got {}",
265 + decoded.len()
266 + )));
267 + }
268 + Ok(())
269 + }
270 +
251 271 /// Validate a developer-defined SDK key. Opaque string identifying which
252 272 /// workspace/org/end-user a JWT session belongs to. Rejects empty, oversize,
253 273 /// null bytes, and control characters — same character rules as `validate_sync_row_id`
M server/tests/harness/storage.rs +9 -2
@@ -153,7 +153,7 @@ impl StorageBackend for InMemoryStorage {
153 153 Ok(upload_id)
154 154 }
155 155
156 - async fn presign_upload_part(&self, s3_key: &S3Key, upload_id: &str, part_number: i32, _expiry_secs: Option<u64>, _max_bytes: Option<i64>) -> Result<String> {
156 + async fn presign_upload_part(&self, s3_key: &S3Key, upload_id: &str, part_number: i32, _expiry_secs: Option<u64>, _max_bytes: Option<i64>, checksum_sha256: Option<&str>) -> Result<String> {
157 157 // Mirror the production range check so a bad part number fails in tests
158 158 // the same way it would against S3.
159 159 if !(1..=s3_storage::MULTIPART_MAX_PARTS as i32).contains(&part_number) {
@@ -162,7 +162,14 @@ impl StorageBackend for InMemoryStorage {
162 162 s3_storage::MULTIPART_MAX_PARTS
163 163 )));
164 164 }
165 - Ok(format!("http://test-storage/{s3_key}?uploadId={upload_id}&partNumber={part_number}"))
165 + // Echo the bound checksum into the URL so tests can assert it reached
166 + // the signer, standing in for the SignedHeaders a real presign carries.
167 + let checksum = checksum_sha256
168 + .map(|c| format!("&checksum={c}"))
169 + .unwrap_or_default();
170 + Ok(format!(
171 + "http://test-storage/{s3_key}?uploadId={upload_id}&partNumber={part_number}{checksum}"
172 + ))
166 173 }
167 174
168 175 async fn complete_multipart_upload(&self, _s3_key: &S3Key, upload_id: &str, parts: &[(i32, String)]) -> Result<()> {
M server/tests/workflows/synckit_blob_multipart.rs +72
@@ -223,6 +223,78 @@ async fn multipart_parts_signs_each_part_with_its_exact_length() {
223 223 }
224 224
225 225 #[tokio::test]
226 + async fn multipart_parts_binds_the_client_checksum_into_the_signed_url() {
227 + // A checksum bound into the presigned URL is one S3 enforces: it rehashes
228 + // the part and rejects a mismatch before the bytes are durable. The client
229 + // can only supply one for a part it has already sealed, which is why it
230 + // asks for a single part at a time.
231 + use base64::Engine;
232 +
233 + let (mut h, _blobs) = harness_with_blobs().await;
234 + let user_id = h.signup("mp_sum", "mp_sum@example.com", "Password1!").await;
235 + let (app_id, _key) = create_internal_app(&h.db, user_id).await;
236 + seed_subscription(&h.db, user_id, app_id, "active", SYNCKIT_MAX_MULTIPART_BLOB_SIZE_BYTES).await;
237 + auth_as(&mut h, user_id, app_id, "user-key");
238 +
239 + let hash = fake_hash(0x18);
240 + let start: Value = h
241 + .client
242 + .post_json(
243 + "/api/sync/blobs/multipart/start",
244 + &json!({ "hash": hash, "size_bytes": LARGE_BLOB }).to_string(),
245 + )
246 + .await
247 + .json();
248 + let upload_id = start["upload_id"].as_str().unwrap().to_string();
249 +
250 + let checksum = base64::engine::general_purpose::STANDARD.encode([7u8; 32]);
251 + let parts: Value = h
252 + .client
253 + .post_json(
254 + "/api/sync/blobs/multipart/parts",
255 + &json!({
256 + "hash": hash, "upload_id": upload_id, "size_bytes": LARGE_BLOB,
257 + "first_part": 3, "count": 1, "checksums": [checksum],
258 + })
259 + .to_string(),
260 + )
261 + .await
262 + .json();
263 + let url = parts["parts"][0]["url"].as_str().unwrap();
264 + assert!(
265 + url.contains(&format!("checksum={checksum}")),
266 + "the checksum must reach the signer, got: {url}"
267 + );
268 + assert_eq!(parts["parts"][0]["part_number"].as_u64().unwrap(), 3);
269 +
270 + // Malformed or misaligned checksums are refused rather than signed: a
271 + // checksum bound to the wrong part would reject a correct upload.
272 + let bad = [
273 + json!({ "first_part": 1, "count": 1, "checksums": ["not base64!!"] }),
274 + json!({ "first_part": 1, "count": 1, "checksums": [base64::engine::general_purpose::STANDARD.encode([7u8; 16])] }),
275 + json!({ "first_part": 1, "count": 2, "checksums": [checksum] }),
276 + json!({ "first_part": 1, "count": 1, "checksums": [checksum, checksum] }),
277 + ];
278 + for case in bad {
279 + let mut body = json!({
280 + "hash": hash, "upload_id": upload_id, "size_bytes": LARGE_BLOB,
281 + });
282 + for (k, v) in case.as_object().unwrap() {
283 + body[k] = v.clone();
284 + }
285 + let resp = h
286 + .client
287 + .post_json("/api/sync/blobs/multipart/parts", &body.to_string())
288 + .await;
289 + assert!(
290 + resp.status.is_client_error(),
291 + "{case} must be refused: {}",
292 + resp.text
293 + );
294 + }
295 + }
296 +
297 + #[tokio::test]
226 298 async fn multipart_complete_then_confirm_records_the_blob() {
227 299 // The end-to-end handoff. Confirm is untouched by multipart: it reads the
228 300 // real object size and charges that, not anything the session declared.
M shared/s3-storage/src/lib.rs +78 -1
@@ -683,6 +683,14 @@ impl S3Client {
683 683 /// defense-in-depth (not a hard, server-enforced ceiling) as
684 684 /// [`Self::presign_upload`] — see its docs for why the authoritative size
685 685 /// check still lives at confirm time.
686 + ///
687 + /// `checksum_sha256` (base64 of the raw 32-byte digest) is signed as
688 + /// `x-amz-checksum-sha256`, and unlike the length this one S3 *does*
689 + /// enforce: it hashes the received part and rejects a mismatch with
690 + /// `BadDigest` before the bytes are durable. The caller must therefore send
691 + /// the header — it is in `SignedHeaders`, so omitting it fails the
692 + /// signature. This is transport integrity (the bytes S3 wrote are the bytes
693 + /// the client hashed), not a statement about what those bytes mean.
686 694 pub async fn presign_upload_part(
687 695 &self,
688 696 key: &str,
@@ -690,6 +698,7 @@ impl S3Client {
690 698 part_number: i32,
691 699 expiry_secs: u64,
692 700 max_bytes: Option<i64>,
701 + checksum_sha256: Option<&str>,
693 702 ) -> Result<String, String> {
694 703 if !(1..=MULTIPART_MAX_PARTS as i32).contains(&part_number) {
695 704 return Err(format!(
@@ -712,6 +721,9 @@ impl S3Client {
712 721 if let Some(n) = max_bytes {
713 722 req = req.content_length(n);
714 723 }
724 + if let Some(c) = checksum_sha256 {
725 + req = req.checksum_sha256(c);
726 + }
715 727
716 728 let presigned = req
717 729 .presigned(presigning_config)
@@ -1254,6 +1266,44 @@ mod tests {
1254 1266 assert!(err.contains("non-empty"), "unexpected error: {err}");
1255 1267 }
1256 1268
1269 + /// The `X-Amz-SignedHeaders` list from a presigned URL.
1270 + fn signed_headers(url: &str) -> String {
1271 + url.split('&')
1272 + .find_map(|p| p.strip_prefix("X-Amz-SignedHeaders="))
1273 + .map(|v| v.replace("%3B", ";"))
1274 + .expect("presigned URL must carry X-Amz-SignedHeaders")
1275 + }
1276 +
1277 + #[tokio::test]
1278 + async fn presign_upload_signs_content_length_when_bound() {
1279 + // Callers rely on `max_bytes` being enforced, and it is enforced only
1280 + // because it lands in SignedHeaders: a client sending a different
1281 + // Content-Length then fails the signature. That also makes the declared
1282 + // size a hard contract — a caller that declares anything other than the
1283 + // exact body length breaks every upload — so pin it here rather than
1284 + // discovering it against production S3.
1285 + let client = test_client();
1286 +
1287 + let bound = client
1288 + .presign_upload("k", "application/octet-stream", 900, None, Some(12_345))
1289 + .await
1290 + .unwrap();
1291 + let headers = signed_headers(&bound);
1292 + assert!(
1293 + headers.contains("content-length"),
1294 + "max_bytes must be signed, got: {headers}"
1295 + );
1296 +
1297 + let unbound = client
1298 + .presign_upload("k", "application/octet-stream", 900, None, None)
1299 + .await
1300 + .unwrap();
1301 + assert!(
1302 + !signed_headers(&unbound).contains("content-length"),
1303 + "without max_bytes the client is free to send any length"
1304 + );
1305 + }
1306 +
1257 1307 #[tokio::test]
1258 1308 async fn presign_upload_part_rejects_out_of_range_part_number() {
1259 1309 // Pre-flight range check: fires before any network call, so the
@@ -1261,7 +1311,7 @@ mod tests {
1261 1311 let client = test_client();
1262 1312 for bad in [0, MULTIPART_MAX_PARTS as i32 + 1] {
1263 1313 let err = client
1264 - .presign_upload_part("k", "uid", bad, 3600, None)
1314 + .presign_upload_part("k", "uid", bad, 3600, None, None)
1265 1315 .await
1266 1316 .expect_err("out-of-range part number must be rejected");
1267 1317 assert!(err.contains("out of range"), "unexpected error: {err}");
@@ -1269,6 +1319,33 @@ mod tests {
1269 1319 }
1270 1320
1271 1321 #[tokio::test]
1322 + async fn presign_upload_part_signs_the_checksum_when_bound() {
1323 + // S3 enforces a bound checksum by rehashing the part, but only if the
1324 + // client sends the header — which it must, because signing it makes it
1325 + // mandatory. Both halves of that live in SignedHeaders.
1326 + let client = test_client();
1327 +
1328 + let bound = client
1329 + .presign_upload_part("k", "uid", 1, 900, Some(64), Some("Zm9vYmFyYmF6"))
1330 + .await
1331 + .unwrap();
1332 + let headers = signed_headers(&bound);
1333 + assert!(
1334 + headers.contains("x-amz-checksum-sha256"),
1335 + "a bound checksum must be signed, got: {headers}"
1336 + );
1337 +
1338 + let unbound = client
1339 + .presign_upload_part("k", "uid", 1, 900, Some(64), None)
1340 + .await
1341 + .unwrap();
1342 + assert!(
1343 + !signed_headers(&unbound).contains("checksum"),
1344 + "no checksum bound means no checksum header is required"
1345 + );
1346 + }
1347 +
1348 + #[tokio::test]
1272 1349 async fn complete_multipart_rejects_empty_parts() {
1273 1350 let client = test_client();
1274 1351 let err = client
M shared/synckit-client/src/client/blob.rs +51 -44
@@ -5,9 +5,9 @@
5 5 //! downloads are re-hashed on arrival and discarded on any mismatch, so a
6 6 //! server that swaps or rolls back a blob cannot slip it past the client.
7 7
8 - use std::collections::VecDeque;
9 8 use std::path::Path;
10 9
10 + use base64::Engine;
11 11 use bytes::{Buf, Bytes, BytesMut};
12 12 use sha2::{Digest, Sha256};
13 13 use tokio::io::AsyncReadExt;
@@ -28,18 +28,19 @@ use super::helpers::{check_response, Idempotency};
28 28 /// Generous (4 GiB) so legitimate large media still flow.
29 29 const MAX_BLOB_BYTES: usize = 4 * 1024 * 1024 * 1024;
30 30
31 - /// How many presigned part URLs [`SyncKitClient::blob_upload_streaming`] pulls
32 - /// per request. Must not exceed the server's own window bound
33 - /// (`BLOB_MULTIPART_PART_URL_WINDOW`), which rejects a larger `count` outright.
34 - /// Pulling as the upload progresses keeps the number of live upload credentials
35 - /// small for a transfer that may never finish.
36 - const PART_URL_WINDOW: u32 = 100;
37 -
38 31 impl SyncKitClient {
39 32 /// Request a presigned upload URL for a blob.
40 33 /// Returns (upload_url, already_exists). If `already_exists` is true,
41 34 /// the blob is already on the server and no upload is needed.
42 35 ///
36 + /// `size_bytes` is the **plaintext** length, the size the caller sees on
37 + /// disk. What [`Self::blob_upload`] actually PUTs is the sealed ciphertext,
38 + /// which is longer by the v3 framing, and the server signs the declared
39 + /// size into the presigned URL as `Content-Length` (a SignedHeader) — so
40 + /// declaring the plaintext length makes every PUT fail the signature. This
41 + /// method converts, keeping the wire-format arithmetic inside the SDK
42 + /// instead of asking three consumer apps to know it.
43 + ///
43 44 /// `already_exists` is the server's claim, not a verified fact. See
44 45 /// [`BlobUploadUrlResponse::already_exists`] for what a false `true` costs
45 46 /// (a withheld blob, never a substituted one).
@@ -49,16 +50,14 @@ impl SyncKitClient {
49 50 hash: &str,
50 51 size_bytes: i64,
51 52 ) -> Result<BlobUploadUrlResponse> {
52 - if size_bytes < 0 {
53 - return Err(SyncKitError::InvalidArgument(
54 - "size_bytes must be non-negative".into(),
55 - ));
56 - }
53 + let plaintext_len = usize::try_from(size_bytes).map_err(|_| {
54 + SyncKitError::InvalidArgument("size_bytes must be non-negative".into())
55 + })?;
57 56 let token = self.require_token()?;
58 57
59 58 let body = Bytes::from(serde_json::to_vec(&BlobUploadUrlRequest {
60 59 hash: hash.to_string(),
61 - size_bytes,
60 + size_bytes: crypto::blob_encrypted_len(plaintext_len) as i64,
62 61 })?);
63 62
64 63 self.retry_request_json(Idempotency::ReadOnly, || {
@@ -89,7 +88,7 @@ impl SyncKitClient {
89 88 pub async fn blob_upload(&self, hash: &str, presigned_url: &str, data: Vec<u8>) -> Result<()> {
90 89 if data.len() > MAX_BLOB_BYTES {
91 90 return Err(SyncKitError::InvalidArgument(format!(
92 - "blob is {} bytes, over the {MAX_BLOB_BYTES}-byte single-upload cap; larger blobs need the (deferred) multipart path",
91 + "blob is {} bytes, over the {MAX_BLOB_BYTES}-byte in-memory cap; use blob_upload_streaming for a file this size",
93 92 data.len()
94 93 )));
95 94 }
@@ -256,7 +255,6 @@ impl SyncKitClient {
256 255 let mut staged = BytesMut::with_capacity(part_size + crypto::BLOB_CHUNK_SIZE);
257 256 staged.extend_from_slice(&crypto::blob_header_bytes(plaintext_len));
258 257
259 - let mut urls: VecDeque<BlobMultipartPartUrl> = VecDeque::new();
260 258 let mut completed = Vec::with_capacity(start.part_count as usize);
261 259 let mut next_part = 1u32;
262 260
@@ -283,7 +281,7 @@ impl SyncKitClient {
283 281 while staged.len() >= part_size && next_part < start.part_count {
284 282 let body = staged.split_to(part_size).freeze();
285 283 let etag = self
286 - .upload_blob_part(hash, start, size_bytes, next_part, &mut urls, body)
284 + .upload_blob_part(hash, start, size_bytes, next_part, body)
287 285 .await?;
288 286 completed.push(BlobMultipartCompletedPart {
289 287 part_number: next_part as i32,
@@ -312,7 +310,7 @@ impl SyncKitClient {
312 310 }
313 311 let body = staged.split().freeze();
314 312 let etag = self
315 - .upload_blob_part(hash, start, size_bytes, next_part, &mut urls, body)
313 + .upload_blob_part(hash, start, size_bytes, next_part, body)
316 314 .await?;
317 315 completed.push(BlobMultipartCompletedPart {
318 316 part_number: next_part as i32,
@@ -322,41 +320,47 @@ impl SyncKitClient {
322 320 Ok(completed)
323 321 }
324 322
325 - /// PUT one part to its presigned URL, refilling the URL window as needed,
326 - /// and return the ETag S3 assigned it.
323 + /// PUT one part to its presigned URL and return the ETag S3 assigned it.
324 + ///
325 + /// The URL is requested for this part alone, and carries the part's SHA-256
326 + /// bound as `x-amz-checksum-sha256`, so S3 rehashes what it receives and
327 + /// rejects a corrupted part at write time instead of letting it sit until a
328 + /// download fails to decrypt. That is the reason parts are not requested in
329 + /// windows: a digest only exists once the part has been sealed, and only one
330 + /// part is ever in memory. The cost is one small round trip per part.
327 331 async fn upload_blob_part(
328 332 &self,
329 333 hash: &str,
330 334 start: &BlobMultipartStartResponse,
331 335 size_bytes: i64,
332 336 part_number: u32,
333 - urls: &mut VecDeque<BlobMultipartPartUrl>,
334 337 body: Bytes,
335 338 ) -> Result<String> {
336 - if urls.is_empty() {
337 - let token = self.require_token()?;
338 - let req_body = Bytes::from(serde_json::to_vec(&BlobMultipartPartsRequest {
339 - hash: hash.to_string(),
340 - upload_id: start.upload_id.clone(),
341 - size_bytes,
342 - first_part: part_number,
343 - count: PART_URL_WINDOW.min(start.part_count - part_number + 1),
344 - })?);
345 - let window: BlobMultipartPartsResponse = self
346 - .retry_request_json(Idempotency::ReadOnly, || {
347 - let req = self
348 - .http
349 - .post(&self.endpoints.blobs_multipart_parts)
350 - .bearer_auth(&token)
351 - .header("content-type", "application/json")
352 - .body(req_body.clone());
353 - async move { check_response(req.send().await?).await }
354 - })
355 - .await?;
356 - urls.extend(window.parts);
357 - }
339 + let checksum =
340 + base64::engine::general_purpose::STANDARD.encode(Sha256::digest(&body));
341 +
342 + let token = self.require_token()?;
343 + let req_body = Bytes::from(serde_json::to_vec(&BlobMultipartPartsRequest {
344 + hash: hash.to_string(),
345 + upload_id: start.upload_id.clone(),
346 + size_bytes,
347 + first_part: part_number,
348 + count: 1,
349 + checksums: vec![checksum.clone()],
350 + })?);
351 + let window: BlobMultipartPartsResponse = self
352 + .retry_request_json(Idempotency::ReadOnly, || {
353 + let req = self
354 + .http
355 + .post(&self.endpoints.blobs_multipart_parts)
356 + .bearer_auth(&token)
357 + .header("content-type", "application/json")
358 + .body(req_body.clone());
359 + async move { check_response(req.send().await?).await }
360 + })
361 + .await?;
358 362
359 - let part = urls.pop_front().ok_or_else(|| {
363 + let part = window.parts.into_iter().next().ok_or_else(|| {
360 364 SyncKitError::Internal(format!("server returned no URL for part {part_number}"))
361 365 })?;
362 366 // The two sides derive the plan independently, so a disagreement means
@@ -379,6 +383,9 @@ impl SyncKitClient {
379 383 .http_stream
380 384 .put(&part.url)
381 385 .header("content-type", "application/octet-stream")
386 + // Mandatory, not optional: the checksum is a signed
387 + // header, so omitting it fails the signature.
388 + .header("x-amz-checksum-sha256", &checksum)
382 389 .body(body.clone());
383 390 async move { check_response(req.send().await?).await }
384 391 },
M shared/synckit-client/src/types.rs +5
@@ -560,6 +560,11 @@ pub(crate) struct BlobMultipartPartsRequest {
560 560 pub size_bytes: i64,
561 561 pub first_part: u32,
562 562 pub count: u32,
563 + /// SHA-256 of each requested part, base64 of the raw digest, aligned with
564 + /// `first_part..first_part + count`. The server binds these into the
565 + /// presigned URLs as `x-amz-checksum-sha256`, so S3 rehashes each part and
566 + /// refuses a corrupted one at write time.
567 + pub checksums: Vec<String>,
563 568 }
564 569
565 570 #[derive(Deserialize)]
M shared/synckit-client/tests/integration.rs +143 -17
@@ -460,6 +460,75 @@ async fn blob_upload_url_success() {
460 460 }
461 461
462 462 #[tokio::test]
463 + async fn blob_upload_url_declares_the_length_the_put_will_carry() {
464 + // The server signs the declared size into the presigned URL as
465 + // Content-Length, a SignedHeader — so declaring anything other than the
466 + // exact ciphertext length makes the PUT fail SigV4. The caller passes the
467 + // plaintext size it sees on disk; the SDK converts. This test pins the two
468 + // halves together, which is the only place the mismatch would show up:
469 + // wiremock does not verify signatures, and the server's own tests use an
470 + // in-memory backend that does not sign at all.
471 + let server = MockServer::start().await;
472 +
473 + let upload_path = "/s3/sized-upload";
474 + Mock::given(method("POST"))
475 + .and(path("/api/v1/sync/blobs/upload"))
476 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
477 + "upload_url": format!("{}{upload_path}", server.uri()),
478 + "already_exists": false,
479 + })))
480 + .mount(&server)
481 + .await;
482 + Mock::given(method("PUT"))
483 + .and(path(upload_path))
484 + .respond_with(ResponseTemplate::new(200))
485 + .mount(&server)
486 + .await;
487 +
488 + let client = authed_client(&server);
489 + client.set_master_key_raw(synckit_client::crypto::generate_master_key());
490 +
491 + // Spans two chunks, so the framing overhead is more than a single chunk's.
492 + let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE + 500))
493 + .map(|i| i as u8)
494 + .collect();
495 + let hash = format!("{:x}", sha2::Sha256::digest(&plaintext));
496 +
497 + let resp = client
498 + .blob_upload_url(&hash, plaintext.len() as i64)
499 + .await
500 + .unwrap();
501 + client
502 + .blob_upload(&hash, &resp.upload_url, plaintext.clone())
503 + .await
504 + .unwrap();
505 +
506 + let reqs = server.received_requests().await.unwrap();
507 + let declared: serde_json::Value = reqs
508 + .iter()
509 + .find(|r| r.url.path() == "/api/v1/sync/blobs/upload")
510 + .unwrap()
511 + .body_json()
512 + .unwrap();
513 + let put_len = reqs
514 + .iter()
515 + .find(|r| r.url.path() == upload_path)
516 + .unwrap()
517 + .body
518 + .len();
519 +
520 + assert_eq!(
521 + declared["size_bytes"].as_u64().unwrap(),
522 + put_len as u64,
523 + "the declared size must equal the bytes actually PUT, or the signature fails"
524 + );
525 + assert!(
526 + put_len > plaintext.len(),
527 + "the PUT carries ciphertext, which is longer than the plaintext"
528 + );
529 + }
530 +
531 + #[tokio::test]
463 532 async fn blob_upload_encrypts_data() {
464 533 let server = MockServer::start().await;
465 534
@@ -3155,24 +3224,45 @@ mod blob_multipart {
3155 3224 p
3156 3225 }
3157 3226
3158 - /// Mount the whole session: start (with the given plan), a parts window
3159 - /// whose URLs all point at one PUT path, and complete.
3227 + /// Stands in for the server's part-URL minting: answers whatever window the
3228 + /// client asked for, rather than a fixed list, since the client requests one
3229 + /// part at a time (it can only checksum a part it has already sealed).
3230 + struct PartsResponder {
3231 + cipher_len: usize,
3232 + part_size: usize,
3233 + base: String,
3234 + }
3235 +
3236 + impl wiremock::Respond for PartsResponder {
3237 + fn respond(&self, req: &wiremock::Request) -> ResponseTemplate {
3238 + let body: serde_json::Value = serde_json::from_slice(&req.body).unwrap();
3239 + let first = body["first_part"].as_u64().unwrap() as usize;
3240 + let count = body["count"].as_u64().unwrap() as usize;
3241 + let part_count = self.cipher_len.div_ceil(self.part_size);
3242 + let last = (first + count - 1).min(part_count);
3243 +
3244 + let parts: Vec<serde_json::Value> = (first..=last)
3245 + .map(|n| {
3246 + let content_length = if n == part_count {
3247 + self.cipher_len - self.part_size * (part_count - 1)
3248 + } else {
3249 + self.part_size
3250 + };
3251 + json!({
3252 + "part_number": n,
3253 + "content_length": content_length,
3254 + "url": format!("{}{PART_PUT_PATH}?partNumber={n}", self.base),
3255 + })
3256 + })
3257 + .collect();
3258 + ResponseTemplate::new(200).set_body_json(json!({ "parts": parts }))
3259 + }
3260 + }
3261 +
3262 + /// Mount the whole session: start (with the given plan), part-URL minting,
3263 + /// the PUT target, and complete.
3160 3264 async fn mount_session(server: &MockServer, cipher_len: usize, part_size: usize) -> u32 {
3161 3265 let part_count = cipher_len.div_ceil(part_size) as u32;
3162 - let parts: Vec<serde_json::Value> = (1..=part_count)
3163 - .map(|n| {
3164 - let content_length = if n == part_count {
3165 - cipher_len - part_size * (part_count as usize - 1)
3166 - } else {
3167 - part_size
3168 - };
3169 - json!({
3170 - "part_number": n,
3171 - "content_length": content_length,
3172 - "url": format!("{}{PART_PUT_PATH}?partNumber={n}", server.uri()),
3173 - })
3174 - })
3175 - .collect();
3176 3266
3177 3267 Mock::given(method("POST"))
3178 3268 .and(path(START_PATH))
@@ -3186,7 +3276,11 @@ mod blob_multipart {
3186 3276 .await;
3187 3277 Mock::given(method("POST"))
3188 3278 .and(path(PARTS_PATH))
3189 - .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "parts": parts })))
3279 + .respond_with(PartsResponder {
3280 + cipher_len,
3281 + part_size,
3282 + base: server.uri(),
3283 + })
3190 3284 .mount(server)
3191 3285 .await;
3192 3286 Mock::given(method("PUT"))
@@ -3254,6 +3348,38 @@ mod blob_multipart {
3254 3348 assert_eq!(put.body.len(), expected, "part {} length", i + 1);
3255 3349 }
3256 3350
3351 + // Each part was requested with the SHA-256 of exactly the bytes that
3352 + // part then carried, which is what S3 rehashes against at write time.
3353 + // The pairing is what matters: a checksum bound to the wrong part is
3354 + // worse than none, since it would reject a correct upload.
3355 + let part_reqs: Vec<&wiremock::Request> =
3356 + reqs.iter().filter(|r| r.url.path() == PARTS_PATH).collect();
3357 + assert_eq!(
3358 + part_reqs.len() as u32,
3359 + part_count,
3360 + "one URL request per part: a digest exists only once the part is sealed"
3361 + );
3362 + for (i, req) in part_reqs.iter().enumerate() {
3363 + let body: serde_json::Value = req.body_json().unwrap();
3364 + assert_eq!(body["first_part"].as_u64().unwrap(), i as u64 + 1);
3365 + assert_eq!(body["count"].as_u64().unwrap(), 1);
3366 + let declared = body["checksums"][0].as_str().unwrap();
3367 + let expected = base64::engine::general_purpose::STANDARD
3368 + .encode(sha2::Sha256::digest(&puts[i].body));
3369 + assert_eq!(declared, expected, "part {} checksum must match its bytes", i + 1);
3370 + // And the client must actually send it: it is a signed header, so
3371 + // dropping it would fail SigV4 at S3.
3372 + assert_eq!(
3373 + puts[i]
3374 + .headers
3375 + .get("x-amz-checksum-sha256")
3376 + .expect("the PUT must carry the checksum header")
3377 + .to_str()
3378 + .unwrap(),
3379 + declared
3380 + );
3381 + }
3382 +
3257 3383 // The concatenated parts are a valid v3 blob for this content address —
3258 3384 // proof that streaming produced the same wire format as the in-memory
3259 3385 // encrypt, boundaries and all.