Skip to main content

max / makenotwork

Add multipart blob sessions for SyncKit Blobs above SYNCKIT_MAX_BLOB_SIZE_BYTES had no upload path: the one-shot presigned PUT refuses them and nothing else existed. Adds the chunked transport, the synckit half of the large-upload plan. Server: /api/v1/sync/blobs/multipart/{start,parts,complete,abort}, mirroring the creator-media session. The blob key is {app_id}/{user_id}/{hash}, derived from the caller's JWT, so ownership is structural and needs no pending_uploads lookup. Part URLs are minted in bounded windows, each with its exact signed Content-Length. Confirm is unchanged and still reads the authoritative object size, but is now bounded by the multipart ceiling, since it cannot tell which transport wrote the object and the one-shot route is already bounded at presign time. The multipart ceiling is the per-user-per-app storage allowance: a blob above it could never be stored, so the session refuses it before staging parts. Client crypto: blob_header_bytes, seal_blob_chunk, blob_encrypted_len, and blob_chunk_count_for, with encrypt_blob_chunked routed through them so the streaming and whole-buffer paths cannot drift. blob_encrypted_len is what a streaming uploader declares to start, so both sides derive identical part boundaries with no round trip. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-21 14:42 UTC
Commit: 7b530acdc24555144bc691900f290c45a3333e03
Parent: d85d7d0
8 files changed, +860 insertions, -49 deletions
M server/src/constants.rs +11
@@ -90,7 +90,14 @@ pub const SYNCKIT_PULL_PAGE_SIZE: i64 = 500;
90 90 pub const SYNCKIT_API_KEY_LENGTH: usize = 32; // 32 bytes = 64 hex chars
91 91 pub const SYNC_LOG_RETAIN_DAYS: i64 = 90;
92 92 pub const SYNC_LOG_COMPACT_MIN_AGE_DAYS: i64 = 7; // Safety margin for cursor-based compaction
93 + /// Largest blob the one-shot presigned PUT will sign. A single PUT holds the
94 + /// whole object in one request the client cannot resume, so this stays modest;
95 + /// bigger blobs go through the multipart session below.
93 96 pub const SYNCKIT_MAX_BLOB_SIZE_BYTES: i64 = 500 * 1024 * 1024; // 500 MB
97 + /// Largest blob the multipart session will accept. Set to the per-user-per-app
98 + /// storage allowance: a blob above it could never be stored anyway, so the
99 + /// session refuses it before it stages S3 parts that confirm would only reject.
100 + pub const SYNCKIT_MAX_MULTIPART_BLOB_SIZE_BYTES: i64 = SYNCKIT_MAX_BLOB_STORAGE_BYTES;
94 101 pub const SYNCKIT_MAX_BLOB_STORAGE_BYTES: i64 = 10 * 1024 * 1024 * 1024; // 10 GB per user per app
95 102 pub const SYNCKIT_MAX_DEVICES_PER_APP: i64 = 50; // Max devices per user per app
96 103 pub const SYNCKIT_BLOB_PRESIGN_EXPIRY_SECS: u64 = 3600; // 1 hour
@@ -520,6 +527,10 @@ const _: () = assert!(SCAN_ZIP_MAX_DEPTH > 0);
520 527 const _: () = assert!(SYNCKIT_PUSH_MAX_CHANGES > 0);
521 528 const _: () = assert!(SYNCKIT_PULL_PAGE_SIZE > 0);
522 529 const _: () = assert!(SYNCKIT_MAX_BLOB_SIZE_BYTES > 0);
530 + // Multipart exists to exceed the one-shot ceiling, and nothing above the
531 + // storage allowance is storable.
532 + const _: () = assert!(SYNCKIT_MAX_MULTIPART_BLOB_SIZE_BYTES > SYNCKIT_MAX_BLOB_SIZE_BYTES);
533 + const _: () = assert!(SYNCKIT_MAX_MULTIPART_BLOB_SIZE_BYTES <= SYNCKIT_MAX_BLOB_STORAGE_BYTES);
523 534 const _: () = assert!(SYNCKIT_JWT_EXPIRY_SECS > 0);
524 535
525 536 // TOTP
M server/src/openapi.rs +12
@@ -54,6 +54,10 @@ use utoipa::OpenApi;
54 54 crate::routes::synckit::sync::cancel_rotation,
55 55 // SyncKit — Blobs
56 56 crate::routes::synckit::blobs::blob_upload_url,
57 + crate::routes::synckit::blobs::blob_multipart_start,
58 + crate::routes::synckit::blobs::blob_multipart_parts,
59 + crate::routes::synckit::blobs::blob_multipart_complete,
60 + crate::routes::synckit::blobs::blob_multipart_abort,
57 61 crate::routes::synckit::blobs::blob_confirm_upload,
58 62 crate::routes::synckit::blobs::blob_download_url,
59 63 ),
@@ -89,6 +93,14 @@ use utoipa::OpenApi;
89 93 crate::routes::synckit::BlobUploadUrlRequest,
90 94 crate::routes::synckit::BlobUploadUrlResponse,
91 95 crate::routes::synckit::BlobConfirmRequest,
96 + crate::routes::synckit::BlobMultipartStartRequest,
97 + crate::routes::synckit::BlobMultipartStartResponse,
98 + crate::routes::synckit::BlobMultipartPartsRequest,
99 + crate::routes::synckit::BlobMultipartPartsResponse,
100 + crate::routes::synckit::BlobMultipartPartUrl,
101 + crate::routes::synckit::BlobMultipartCompleteRequest,
102 + crate::routes::synckit::BlobMultipartCompletedPart,
103 + crate::routes::synckit::BlobMultipartAbortRequest,
92 104 crate::routes::synckit::BlobDownloadUrlRequest,
93 105 crate::routes::synckit::BlobDownloadUrlResponse,
94 106 // SyncKit — Account & subscription
M server/src/routes/synckit/blobs.rs +314 -31
@@ -19,8 +19,10 @@ use crate::{
19 19 };
20 20
21 21 use super::{
22 - BlobConfirmRequest, BlobDownloadUrlRequest, BlobDownloadUrlResponse, BlobUploadUrlRequest,
23 - BlobUploadUrlResponse,
22 + BlobConfirmRequest, BlobDownloadUrlRequest, BlobDownloadUrlResponse,
23 + BlobMultipartAbortRequest, BlobMultipartCompleteRequest, BlobMultipartPartUrl,
24 + BlobMultipartPartsRequest, BlobMultipartPartsResponse, BlobMultipartStartRequest,
25 + BlobMultipartStartResponse, BlobUploadUrlRequest, BlobUploadUrlResponse,
24 26 };
25 27
26 28 /// Request a pre-signed S3 upload URL for a blob.
@@ -44,40 +46,29 @@ pub(super) async fn blob_upload_url(
44 46 .as_ref()
45 47 .ok_or_else(|| AppError::ServiceUnavailable("SyncKit blob storage is not configured".to_string()))?;
46 48
47 - validation::validate_sync_blob_hash(&req.hash)?;
49 + // The one-shot ceiling, not the multipart one: a single PUT is one
50 + // unresumable request, so anything larger belongs on the multipart session.
48 51 if req.size_bytes <= 0 || req.size_bytes > constants::SYNCKIT_MAX_BLOB_SIZE_BYTES {
49 52 return Err(AppError::BadRequest(format!(
50 - "Blob size must be between 1 and {} bytes",
53 + "Blob size must be between 1 and {} bytes; larger blobs upload through the multipart session",
51 54 constants::SYNCKIT_MAX_BLOB_SIZE_BYTES
52 55 )));
53 56 }
54 57
55 - // Paid-only gate for first-party apps: don't even hand out an upload URL to
56 - // an unsubscribed user, so they can't stage orphan S3 objects. Non-internal
57 - // (developer-billed) apps are always allowed here. Storage-cap enforcement
58 - // still happens atomically at confirm time.
59 - if !db::synckit::internal_write_allowed(&db, sync_user.app_id, sync_user.user_id).await? {
60 - return Ok((
61 - StatusCode::PAYMENT_REQUIRED,
62 - Json(json!({ "reason": "no_subscription" })),
63 - )
64 - .into_response());
65 - }
66 -
67 - // Check dedup — if this hash already exists, skip upload
68 - if let Some(_existing) = db::synckit::get_sync_blob_by_hash(
69 - &db,
70 - sync_user.app_id,
71 - sync_user.user_id,
72 - &req.hash,
73 - )
74 - .await?
75 - {
76 - return Ok(Json(BlobUploadUrlResponse {
77 - upload_url: String::new(),
78 - already_exists: true,
79 - })
80 - .into_response());
58 + // Paid-only gate + content-address dedup. The gate refuses an unsubscribed
59 + // first-party user before any upload credential exists, so they can't stage
60 + // orphan S3 objects; storage-cap enforcement still happens atomically at
61 + // confirm time. Non-internal (developer-billed) apps always pass here.
62 + match blob_upload_gate(&db, &sync_user, &req.hash).await? {
63 + Some(BlobUploadStop::NoSubscription) => return Ok(no_subscription_response()),
64 + Some(BlobUploadStop::AlreadyExists) => {
65 + return Ok(Json(BlobUploadUrlResponse {
66 + upload_url: String::new(),
67 + already_exists: true,
68 + })
69 + .into_response())
70 + }
71 + None => {}
81 72 }
82 73
83 74 let s3_key = crate::storage::S3Client::generate_synckit_blob_key(
@@ -108,6 +99,295 @@ pub(super) async fn blob_upload_url(
108 99 .into_response())
109 100 }
110 101
102 + // ── Multipart blob session (large blobs) ──
103 + //
104 + // The chunked counterpart to `blob_upload_url`, and the only route past
105 + // `SYNCKIT_MAX_BLOB_SIZE_BYTES`: a one-shot presigned PUT is a single
106 + // unresumable request, so it keeps the modest ceiling while multipart carries
107 + // the large blobs. These endpoints replace the *transport* only — the client
108 + // still finishes at `/blobs/confirm`, which reads the authoritative object size
109 + // from S3 and does all the quota/billing work unchanged.
110 + //
111 + // Authorization needs no key lookup here (unlike the creator-media multipart
112 + // session, which signs owner-less `staging/{uuid}` keys): a synckit blob key is
113 + // `{app_id}/{user_id}/{hash}`, derived server-side from the caller's JWT, so a
114 + // caller can only ever address their own blob.
115 +
116 + /// Largest window of presigned part URLs one `parts` call will mint. The client
117 + /// pulls them as it progresses rather than holding hundreds of live
118 + /// credentials for an upload that may never finish.
119 + const BLOB_MULTIPART_PART_URL_WINDOW: u32 = 100;
120 +
121 + /// Why a blob upload must not open. Both transports run the same gate and
122 + /// render it in their own response shape.
123 + enum BlobUploadStop {
124 + /// First-party app, unsubscribed user: don't hand out any upload
125 + /// credential, so they cannot stage orphan S3 objects.
126 + NoSubscription,
127 + /// This content address is already stored for this user/app.
128 + AlreadyExists,
129 + }
130 +
131 + /// Shared pre-upload checks: hash shape, the paid-only gate, and content-address
132 + /// dedup. `Ok(None)` means the caller may open an upload. Size validation stays
133 + /// at the call site, since the two transports have different ceilings.
134 + async fn blob_upload_gate(
135 + db: &PgPool,
136 + sync_user: &SyncUser,
137 + hash: &str,
138 + ) -> Result<Option<BlobUploadStop>> {
139 + validation::validate_sync_blob_hash(hash)?;
140 +
141 + if !db::synckit::internal_write_allowed(db, sync_user.app_id, sync_user.user_id).await? {
142 + return Ok(Some(BlobUploadStop::NoSubscription));
143 + }
144 +
145 + if db::synckit::get_sync_blob_by_hash(db, sync_user.app_id, sync_user.user_id, hash)
146 + .await?
147 + .is_some()
148 + {
149 + return Ok(Some(BlobUploadStop::AlreadyExists));
150 + }
151 +
152 + Ok(None)
153 + }
154 +
155 + /// The 402 both transports return for an unsubscribed first-party user.
156 + fn no_subscription_response() -> Response {
157 + (
158 + StatusCode::PAYMENT_REQUIRED,
159 + Json(json!({ "reason": "no_subscription" })),
160 + )
161 + .into_response()
162 + }
163 +
164 + /// Open a multipart upload session for a large blob.
165 + ///
166 + /// `size_bytes` is the ciphertext length, which the client derives from the
167 + /// plaintext length alone (`blob_encrypted_len`) before sealing anything. The
168 + /// part geometry is pure arithmetic over it, so both sides compute identical
169 + /// boundaries without a round trip.
170 + #[utoipa::path(post, path = "/api/v1/sync/blobs/multipart/start", tag = "SyncKit",
171 + request_body = BlobMultipartStartRequest,
172 + responses((status = 200, description = "Multipart session opened", body = BlobMultipartStartResponse)),
173 + security(("bearer" = [])),
174 + )]
175 + #[tracing::instrument(skip_all, name = "synckit::blob_multipart_start")]
176 + pub(super) async fn blob_multipart_start(
177 + State(db): State<PgPool>,
178 + State(storage): State<crate::AppStorage>,
179 + sync_user: SyncUser,
180 + Json(req): Json<BlobMultipartStartRequest>,
181 + ) -> Result<Response> {
182 + let synckit_s3 = storage.synckit_s3
183 + .as_ref()
184 + .ok_or_else(|| AppError::ServiceUnavailable("SyncKit blob storage is not configured".to_string()))?;
185 +
186 + if req.size_bytes <= 0 || req.size_bytes > constants::SYNCKIT_MAX_MULTIPART_BLOB_SIZE_BYTES {
187 + return Err(AppError::BadRequest(format!(
188 + "Blob size must be between 1 and {} bytes",
189 + constants::SYNCKIT_MAX_MULTIPART_BLOB_SIZE_BYTES
190 + )));
191 + }
192 +
193 + match blob_upload_gate(&db, &sync_user, &req.hash).await? {
194 + Some(BlobUploadStop::NoSubscription) => return Ok(no_subscription_response()),
195 + Some(BlobUploadStop::AlreadyExists) => {
196 + return Ok(Json(BlobMultipartStartResponse {
197 + upload_id: String::new(),
198 + part_size: 0,
199 + part_count: 0,
200 + already_exists: true,
201 + })
202 + .into_response())
203 + }
204 + None => {}
205 + }
206 +
207 + let plan = s3_storage::MultipartPlan::auto(req.size_bytes as u64)
208 + .map_err(AppError::BadRequest)?;
209 +
210 + let s3_key = crate::storage::S3Client::generate_synckit_blob_key(
211 + sync_user.app_id, sync_user.user_id, &req.hash,
212 + );
213 +
214 + // Track the session so the orphan reaper aborts it if the client vanishes.
215 + // An abandoned multipart upload leaves no object at all, only billed parts,
216 + // which the reaper recovers by listing sessions for this key.
217 + db::pending_uploads::record_pending_upload(&db, sync_user.user_id, &s3_key, "synckit").await?;
218 +
219 + let upload_id = synckit_s3
220 + .create_multipart_upload(&s3_key, "application/octet-stream")
221 + .await?;
222 +
223 + tracing::info!(
224 + app = %sync_user.app_id, user = %sync_user.user_id,
225 + size = req.size_bytes, parts = plan.part_count,
226 + "SyncKit multipart blob upload started"
227 + );
228 +
229 + Ok(Json(BlobMultipartStartResponse {
230 + upload_id,
231 + part_size: plan.part_size,
232 + part_count: plan.part_count,
233 + already_exists: false,
234 + })
235 + .into_response())
236 + }
237 +
238 + /// Mint a bounded window of presigned `UploadPart` URLs, each carrying its exact
239 + /// signed `Content-Length` — the same defense-in-depth the one-shot presign
240 + /// applies.
241 + #[utoipa::path(post, path = "/api/v1/sync/blobs/multipart/parts", tag = "SyncKit",
242 + request_body = BlobMultipartPartsRequest,
243 + responses((status = 200, description = "Presigned part URLs", body = BlobMultipartPartsResponse)),
244 + security(("bearer" = [])),
245 + )]
246 + #[tracing::instrument(skip_all, name = "synckit::blob_multipart_parts")]
247 + pub(super) async fn blob_multipart_parts(
248 + State(storage): State<crate::AppStorage>,
249 + sync_user: SyncUser,
250 + Json(req): Json<BlobMultipartPartsRequest>,
251 + ) -> Result<Response> {
252 + let synckit_s3 = storage.synckit_s3
253 + .as_ref()
254 + .ok_or_else(|| AppError::ServiceUnavailable("SyncKit blob storage is not configured".to_string()))?;
255 +
256 + validation::validate_sync_blob_hash(&req.hash)?;
257 + if req.size_bytes <= 0 || req.size_bytes > constants::SYNCKIT_MAX_MULTIPART_BLOB_SIZE_BYTES {
258 + return Err(AppError::BadRequest("Blob size is out of range".to_string()));
259 + }
260 +
261 + let plan = s3_storage::MultipartPlan::auto(req.size_bytes as u64)
262 + .map_err(AppError::BadRequest)?;
263 +
264 + if req.count == 0 || req.count > BLOB_MULTIPART_PART_URL_WINDOW {
265 + return Err(AppError::BadRequest(format!(
266 + "count must be between 1 and {BLOB_MULTIPART_PART_URL_WINDOW}"
267 + )));
268 + }
269 + if req.first_part == 0 || req.first_part > plan.part_count {
270 + return Err(AppError::BadRequest(format!(
271 + "first_part must be between 1 and {}",
272 + plan.part_count
273 + )));
274 + }
275 +
276 + let s3_key = crate::storage::S3Client::generate_synckit_blob_key(
277 + sync_user.app_id, sync_user.user_id, &req.hash,
278 + );
279 +
280 + let expires_in = constants::SYNCKIT_BLOB_PRESIGN_EXPIRY_SECS;
281 + let last = (req.first_part + req.count - 1).min(plan.part_count);
282 + let mut parts = Vec::with_capacity((last - req.first_part + 1) as usize);
283 + for part_number in req.first_part..=last {
284 + let content_length = plan.part_len(part_number);
285 + let url = synckit_s3
286 + .presign_upload_part(
287 + &s3_key,
288 + &req.upload_id,
289 + part_number as i32,
290 + Some(expires_in),
291 + Some(content_length as i64),
292 + )
293 + .await
294 + .context("presign upload part for sync blob")?;
295 + parts.push(BlobMultipartPartUrl {
296 + part_number: part_number as i32,
297 + content_length,
298 + url,
299 + });
300 + }
301 +
302 + Ok(Json(BlobMultipartPartsResponse { parts, expires_in }).into_response())
303 + }
304 +
305 + /// Assemble the uploaded parts into the blob object.
306 + ///
307 + /// Transport only: the client then calls `/blobs/confirm`, which reads the real
308 + /// object size from S3 and applies every quota and billing rule.
309 + #[utoipa::path(post, path = "/api/v1/sync/blobs/multipart/complete", tag = "SyncKit",
310 + request_body = BlobMultipartCompleteRequest,
311 + responses((status = 204, description = "Parts assembled")),
312 + security(("bearer" = [])),
313 + )]
314 + #[tracing::instrument(skip_all, name = "synckit::blob_multipart_complete")]
315 + pub(super) async fn blob_multipart_complete(
316 + State(storage): State<crate::AppStorage>,
317 + sync_user: SyncUser,
318 + Json(req): Json<BlobMultipartCompleteRequest>,
319 + ) -> Result<Response> {
320 + let synckit_s3 = storage.synckit_s3
321 + .as_ref()
322 + .ok_or_else(|| AppError::ServiceUnavailable("SyncKit blob storage is not configured".to_string()))?;
323 +
324 + validation::validate_sync_blob_hash(&req.hash)?;
325 + if req.parts.is_empty() {
326 + return Err(AppError::BadRequest("No parts to complete".to_string()));
327 + }
328 +
329 + let s3_key = crate::storage::S3Client::generate_synckit_blob_key(
330 + sync_user.app_id, sync_user.user_id, &req.hash,
331 + );
332 +
333 + let parts: Vec<(i32, String)> = req
334 + .parts
335 + .into_iter()
336 + .map(|p| (p.part_number, p.etag))
337 + .collect();
338 +
339 + synckit_s3
340 + .complete_multipart_upload(&s3_key, &req.upload_id, &parts)
341 + .await?;
342 +
343 + tracing::info!(
344 + app = %sync_user.app_id, user = %sync_user.user_id, parts = parts.len(),
345 + "SyncKit multipart blob upload completed"
346 + );
347 +
348 + Ok(StatusCode::NO_CONTENT.into_response())
349 + }
350 +
351 + /// Release the parts of an abandoned session (client cancel).
352 + ///
353 + /// Incomplete multipart uploads bill for their parts until aborted, so a client
354 + /// that cleans up on cancel is the cheapest fix; the orphan reaper is the
355 + /// backstop for clients that vanish.
356 + #[utoipa::path(post, path = "/api/v1/sync/blobs/multipart/abort", tag = "SyncKit",
357 + request_body = BlobMultipartAbortRequest,
358 + responses((status = 204, description = "Session aborted")),
359 + security(("bearer" = [])),
360 + )]
361 + #[tracing::instrument(skip_all, name = "synckit::blob_multipart_abort")]
362 + pub(super) async fn blob_multipart_abort(
363 + State(db): State<PgPool>,
364 + State(storage): State<crate::AppStorage>,
365 + sync_user: SyncUser,
366 + Json(req): Json<BlobMultipartAbortRequest>,
367 + ) -> Result<Response> {
368 + let synckit_s3 = storage.synckit_s3
369 + .as_ref()
370 + .ok_or_else(|| AppError::ServiceUnavailable("SyncKit blob storage is not configured".to_string()))?;
371 +
372 + validation::validate_sync_blob_hash(&req.hash)?;
373 +
374 + let s3_key = crate::storage::S3Client::generate_synckit_blob_key(
375 + sync_user.app_id, sync_user.user_id, &req.hash,
376 + );
377 +
378 + synckit_s3.abort_multipart_upload(&s3_key, &req.upload_id).await?;
379 + // The session is gone, so the reaper has nothing left to find; drop the
380 + // tracking row rather than leaving it to age out.
381 + db::pending_uploads::remove_pending_upload(&db, sync_user.user_id, &s3_key, "synckit").await?;
382 +
383 + tracing::info!(
384 + app = %sync_user.app_id, user = %sync_user.user_id,
385 + "SyncKit multipart blob upload aborted"
386 + );
387 +
388 + Ok(StatusCode::NO_CONTENT.into_response())
389 + }
390 +
111 391 /// Confirm that a blob upload to S3 completed successfully.
112 392 ///
113 393 /// Verifies the object exists in S3, then records it in the database.
@@ -159,7 +439,10 @@ pub(super) async fn blob_confirm_upload(
159 439 .ok_or_else(|| {
160 440 AppError::BadRequest("Blob not found in storage — upload before confirming".to_string())
161 441 })?;
162 - if actual_size <= 0 || actual_size > constants::SYNCKIT_MAX_BLOB_SIZE_BYTES {
442 + // Bounded by the multipart ceiling, not the one-shot one: confirm cannot
443 + // tell which transport wrote the object, and the one-shot route is already
444 + // bounded to its own ceiling by the signed `Content-Length` at presign time.
445 + if actual_size <= 0 || actual_size > constants::SYNCKIT_MAX_MULTIPART_BLOB_SIZE_BYTES {
163 446 return Err(AppError::BadRequest(format!(
164 447 "Stored blob size {actual_size} is out of range"
165 448 )));
M server/src/routes/synckit/mod.rs +70
@@ -374,6 +374,68 @@ pub(crate) struct BlobConfirmRequest {
374 374 // is ignored by deserialization (no `deny_unknown_fields`).
375 375 }
376 376
377 + /// Open a multipart blob session. `size_bytes` is the *ciphertext* length,
378 + /// which the client knows before sealing anything (`blob_encrypted_len`), so
379 + /// both sides derive the same part geometry from it without a round trip.
380 + #[derive(Deserialize, utoipa::ToSchema)]
381 + pub(crate) struct BlobMultipartStartRequest {
382 + pub hash: String,
383 + pub size_bytes: i64,
384 + }
385 +
386 + #[derive(Serialize, utoipa::ToSchema)]
387 + pub(crate) struct BlobMultipartStartResponse {
388 + upload_id: String,
389 + part_size: usize,
390 + part_count: u32,
391 + /// Same dedup short-circuit as the one-shot upload: when true no session
392 + /// was opened and the other fields are empty.
393 + already_exists: bool,
394 + }
395 +
396 + #[derive(Deserialize, utoipa::ToSchema)]
397 + pub(crate) struct BlobMultipartPartsRequest {
398 + pub hash: String,
399 + pub upload_id: String,
400 + /// Must match the `size_bytes` passed to `start` — the plan is deterministic
401 + /// in it, and a different value would sign the wrong `Content-Length`s.
402 + pub size_bytes: i64,
403 + pub first_part: u32,
404 + pub count: u32,
405 + }
406 +
407 + #[derive(Serialize, utoipa::ToSchema)]
408 + pub(crate) struct BlobMultipartPartUrl {
409 + part_number: i32,
410 + content_length: u64,
411 + url: String,
412 + }
413 +
414 + #[derive(Serialize, utoipa::ToSchema)]
415 + pub(crate) struct BlobMultipartPartsResponse {
416 + parts: Vec<BlobMultipartPartUrl>,
417 + expires_in: u64,
418 + }
419 +
420 + #[derive(Deserialize, utoipa::ToSchema)]
421 + pub(crate) struct BlobMultipartCompletedPart {
422 + pub part_number: i32,
423 + pub etag: String,
424 + }
425 +
426 + #[derive(Deserialize, utoipa::ToSchema)]
427 + pub(crate) struct BlobMultipartCompleteRequest {
428 + pub hash: String,
429 + pub upload_id: String,
430 + pub parts: Vec<BlobMultipartCompletedPart>,
431 + }
432 +
433 + #[derive(Deserialize, utoipa::ToSchema)]
434 + pub(crate) struct BlobMultipartAbortRequest {
435 + pub hash: String,
436 + pub upload_id: String,
437 + }
438 +
377 439 #[derive(Deserialize, utoipa::ToSchema)]
378 440 pub(crate) struct BlobDownloadUrlRequest {
379 441 pub hash: String,
@@ -605,6 +667,14 @@ pub fn synckit_routes(synckit_jwt_secret: Option<std::sync::Arc<String>>) -> Csr
605 667 .route("/api/v1/sync/keys/rotate/complete", post_csrf_skip(SYNCKIT_JWT_SKIP, sync::complete_rotation))
606 668 .route("/api/sync/blobs/upload", post_csrf_skip(SYNCKIT_JWT_SKIP, blobs::blob_upload_url))
607 669 .route("/api/v1/sync/blobs/upload", post_csrf_skip(SYNCKIT_JWT_SKIP, blobs::blob_upload_url))
670 + .route("/api/sync/blobs/multipart/start", post_csrf_skip(SYNCKIT_JWT_SKIP, blobs::blob_multipart_start))
671 + .route("/api/v1/sync/blobs/multipart/start", post_csrf_skip(SYNCKIT_JWT_SKIP, blobs::blob_multipart_start))
672 + .route("/api/sync/blobs/multipart/parts", post_csrf_skip(SYNCKIT_JWT_SKIP, blobs::blob_multipart_parts))
673 + .route("/api/v1/sync/blobs/multipart/parts", post_csrf_skip(SYNCKIT_JWT_SKIP, blobs::blob_multipart_parts))
674 + .route("/api/sync/blobs/multipart/complete", post_csrf_skip(SYNCKIT_JWT_SKIP, blobs::blob_multipart_complete))
675 + .route("/api/v1/sync/blobs/multipart/complete", post_csrf_skip(SYNCKIT_JWT_SKIP, blobs::blob_multipart_complete))
676 + .route("/api/sync/blobs/multipart/abort", post_csrf_skip(SYNCKIT_JWT_SKIP, blobs::blob_multipart_abort))
677 + .route("/api/v1/sync/blobs/multipart/abort", post_csrf_skip(SYNCKIT_JWT_SKIP, blobs::blob_multipart_abort))
608 678 .route("/api/sync/blobs/confirm", post_csrf_skip(SYNCKIT_JWT_SKIP, blobs::blob_confirm_upload))
609 679 .route("/api/v1/sync/blobs/confirm", post_csrf_skip(SYNCKIT_JWT_SKIP, blobs::blob_confirm_upload))
610 680 .route("/api/sync/blobs/download", post_csrf_skip(SYNCKIT_JWT_SKIP, blobs::blob_download_url))
M server/tests/workflows/mod.rs +1
@@ -20,6 +20,7 @@ mod synckit;
20 20 mod synckit_billing;
21 21 mod synckit_per_key_storage;
22 22 mod synckit_paid_sync;
23 + mod synckit_blob_multipart;
23 24 mod synckit_security;
24 25 mod synckit_adversarial;
25 26 mod pages;
A server/tests/workflows/synckit_blob_multipart.rs +338
@@ -0,0 +1,338 @@
1 + //! SyncKit multipart blob sessions: the transport that carries blobs past the
2 + //! one-shot presigned-PUT ceiling. The session replaces the transport only —
3 + //! the client still finishes at `/blobs/confirm`, which reads the authoritative
4 + //! object size from S3 and applies every quota and billing rule unchanged.
5 +
6 + use serde_json::{json, Value};
7 +
8 + use makenotwork::constants::{
9 + SYNCKIT_MAX_BLOB_SIZE_BYTES, SYNCKIT_MAX_MULTIPART_BLOB_SIZE_BYTES,
10 + };
11 +
12 + use super::synckit_paid_sync::{
13 + auth_as, create_internal_app, fake_hash, harness_with_blobs, seed_subscription,
14 + };
15 +
16 + /// Comfortably above the one-shot ceiling, comfortably below the multipart one.
17 + const LARGE_BLOB: i64 = 600 * 1024 * 1024;
18 + const _: () = assert!(LARGE_BLOB > SYNCKIT_MAX_BLOB_SIZE_BYTES);
19 + const _: () = assert!(LARGE_BLOB < SYNCKIT_MAX_MULTIPART_BLOB_SIZE_BYTES);
20 +
21 + #[tokio::test]
22 + async fn multipart_start_opens_the_band_above_the_single_put_ceiling() {
23 + // Before multipart, a blob over SYNCKIT_MAX_BLOB_SIZE_BYTES had no upload
24 + // path at all: the one-shot presign refuses it and nothing else existed.
25 + let (mut h, _blobs) = harness_with_blobs().await;
26 + let user_id = h.signup("mp_band", "mp_band@example.com", "Password1!").await;
27 + let (app_id, _key) = create_internal_app(&h.db, user_id).await;
28 + seed_subscription(&h.db, user_id, app_id, "active", SYNCKIT_MAX_MULTIPART_BLOB_SIZE_BYTES).await;
29 + auth_as(&mut h, user_id, app_id, "user-key");
30 +
31 + let hash = fake_hash(0x11);
32 + let one_shot = h
33 + .client
34 + .post_json(
35 + "/api/sync/blobs/upload",
36 + &json!({ "hash": hash, "size_bytes": LARGE_BLOB }).to_string(),
37 + )
38 + .await;
39 + assert!(
40 + one_shot.status.is_client_error(),
41 + "the one-shot presign must still refuse a blob over its ceiling: {}",
42 + one_shot.text
43 + );
44 +
45 + let resp = h
46 + .client
47 + .post_json(
48 + "/api/sync/blobs/multipart/start",
49 + &json!({ "hash": hash, "size_bytes": LARGE_BLOB }).to_string(),
50 + )
51 + .await;
52 + assert!(resp.status.is_success(), "multipart start failed: {}", resp.text);
53 +
54 + let body: Value = resp.json();
55 + assert!(!body["already_exists"].as_bool().unwrap());
56 + assert!(!body["upload_id"].as_str().unwrap().is_empty());
57 + let part_size = body["part_size"].as_u64().unwrap();
58 + let part_count = body["part_count"].as_u64().unwrap();
59 + assert!(part_size >= s3_storage::MULTIPART_MIN_PART_SIZE as u64, "part size under S3's floor");
60 + assert!(part_count > 1, "a 600 MB blob must span several parts, got {part_count}");
61 + assert_eq!(
62 + part_count,
63 + (LARGE_BLOB as u64).div_ceil(part_size),
64 + "geometry must be pure arithmetic over the declared size"
65 + );
66 + }
67 +
68 + #[tokio::test]
69 + async fn multipart_start_refuses_a_blob_over_the_multipart_ceiling() {
70 + // Nothing above the per-user storage allowance is storable, so the session
71 + // refuses it before staging parts that confirm could only reject.
72 + let (mut h, blobs) = harness_with_blobs().await;
73 + let user_id = h.signup("mp_huge", "mp_huge@example.com", "Password1!").await;
74 + let (app_id, _key) = create_internal_app(&h.db, user_id).await;
75 + seed_subscription(&h.db, user_id, app_id, "active", SYNCKIT_MAX_MULTIPART_BLOB_SIZE_BYTES).await;
76 + auth_as(&mut h, user_id, app_id, "user-key");
77 +
78 + for size in [0, -1, SYNCKIT_MAX_MULTIPART_BLOB_SIZE_BYTES + 1] {
79 + let resp = h
80 + .client
81 + .post_json(
82 + "/api/sync/blobs/multipart/start",
83 + &json!({ "hash": fake_hash(0x12), "size_bytes": size }).to_string(),
84 + )
85 + .await;
86 + assert!(
87 + resp.status.is_client_error(),
88 + "size {size} must be refused: {}",
89 + resp.text
90 + );
91 + }
92 + assert_eq!(blobs.open_multipart_count(), 0, "a refused start must not stage a session");
93 + }
94 +
95 + #[tokio::test]
96 + async fn multipart_start_refused_without_subscription() {
97 + // Same paid-only gate as the one-shot presign: an unsubscribed first-party
98 + // user gets no upload credential of any shape, so cannot stage orphans.
99 + let (mut h, blobs) = harness_with_blobs().await;
100 + let user_id = h.signup("mp_nosub", "mp_nosub@example.com", "Password1!").await;
101 + let (app_id, _key) = create_internal_app(&h.db, user_id).await;
102 + auth_as(&mut h, user_id, app_id, "user-key");
103 +
104 + let resp = h
105 + .client
106 + .post_json(
107 + "/api/sync/blobs/multipart/start",
108 + &json!({ "hash": fake_hash(0x13), "size_bytes": LARGE_BLOB }).to_string(),
109 + )
110 + .await;
111 + assert_eq!(resp.status, 402, "start must be refused without a sub: {}", resp.text);
112 + assert_eq!(resp.json::<Value>()["reason"], "no_subscription");
113 + assert_eq!(blobs.open_multipart_count(), 0, "a refused start must not stage a session");
114 + }
115 +
116 + #[tokio::test]
117 + async fn multipart_start_short_circuits_on_an_existing_content_address() {
118 + // Dedup runs before the session opens, so re-uploading a blob the user
119 + // already stored costs no S3 state at all.
120 + let (mut h, blobs) = harness_with_blobs().await;
121 + let user_id = h.signup("mp_dedup", "mp_dedup@example.com", "Password1!").await;
122 + let (app_id, _key) = create_internal_app(&h.db, user_id).await;
123 + seed_subscription(&h.db, user_id, app_id, "active", SYNCKIT_MAX_MULTIPART_BLOB_SIZE_BYTES).await;
124 + auth_as(&mut h, user_id, app_id, "user-key");
125 +
126 + let hash = fake_hash(0x14);
127 + blobs.put(&format!("{app_id}/{user_id}/{hash}"), vec![0u8; 1000]);
128 + let confirm = h
129 + .client
130 + .post_json("/api/sync/blobs/confirm", &json!({ "hash": hash }).to_string())
131 + .await;
132 + assert_eq!(confirm.status, 204, "seed confirm failed: {}", confirm.text);
133 +
134 + let resp = h
135 + .client
136 + .post_json(
137 + "/api/sync/blobs/multipart/start",
138 + &json!({ "hash": hash, "size_bytes": LARGE_BLOB }).to_string(),
139 + )
140 + .await;
141 + assert!(resp.status.is_success(), "dedup start failed: {}", resp.text);
142 +
143 + let body: Value = resp.json();
144 + assert!(body["already_exists"].as_bool().unwrap());
145 + assert!(body["upload_id"].as_str().unwrap().is_empty());
146 + assert_eq!(blobs.open_multipart_count(), 0, "dedup must not open a session");
147 + }
148 +
149 + #[tokio::test]
150 + async fn multipart_parts_signs_each_part_with_its_exact_length() {
151 + // Every part URL carries its own signed Content-Length. They must tile the
152 + // declared object exactly: a client that cannot predict the boundaries
153 + // cannot seal chunks into parts without buffering the whole blob.
154 + let (mut h, _blobs) = harness_with_blobs().await;
155 + let user_id = h.signup("mp_parts", "mp_parts@example.com", "Password1!").await;
156 + let (app_id, _key) = create_internal_app(&h.db, user_id).await;
157 + seed_subscription(&h.db, user_id, app_id, "active", SYNCKIT_MAX_MULTIPART_BLOB_SIZE_BYTES).await;
158 + auth_as(&mut h, user_id, app_id, "user-key");
159 +
160 + let hash = fake_hash(0x15);
161 + let start: Value = h
162 + .client
163 + .post_json(
164 + "/api/sync/blobs/multipart/start",
165 + &json!({ "hash": hash, "size_bytes": LARGE_BLOB }).to_string(),
166 + )
167 + .await
168 + .json();
169 + let upload_id = start["upload_id"].as_str().unwrap().to_string();
170 + let part_count = start["part_count"].as_u64().unwrap();
171 + let part_size = start["part_size"].as_u64().unwrap();
172 +
173 + // One window covers this object (part_count < the 100-part window), and the
174 + // response clamps to the real part count rather than over-minting.
175 + let parts: Value = h
176 + .client
177 + .post_json(
178 + "/api/sync/blobs/multipart/parts",
179 + &json!({
180 + "hash": hash, "upload_id": upload_id, "size_bytes": LARGE_BLOB,
181 + "first_part": 1, "count": 100,
182 + })
183 + .to_string(),
184 + )
185 + .await
186 + .json();
187 + let listed = parts["parts"].as_array().unwrap();
188 + assert_eq!(listed.len() as u64, part_count, "window must clamp to the real part count");
189 +
190 + let mut total = 0u64;
191 + for (i, part) in listed.iter().enumerate() {
192 + assert_eq!(part["part_number"].as_u64().unwrap(), i as u64 + 1, "parts are 1-based and ordered");
193 + let len = part["content_length"].as_u64().unwrap();
194 + if (i as u64) < part_count - 1 {
195 + assert_eq!(len, part_size, "every part but the last is a full part");
196 + } else {
197 + assert!(len > 0 && len <= part_size, "final part is the remainder");
198 + }
199 + assert!(!part["url"].as_str().unwrap().is_empty());
200 + total += len;
201 + }
202 + assert_eq!(total, LARGE_BLOB as u64, "signed lengths must tile the object exactly");
203 +
204 + // Out-of-range windows are refused rather than silently clamped.
205 + for (first_part, count) in [(0u32, 1u32), (1, 0), (1, 101), (part_count as u32 + 1, 1)] {
206 + let resp = h
207 + .client
208 + .post_json(
209 + "/api/sync/blobs/multipart/parts",
210 + &json!({
211 + "hash": hash, "upload_id": upload_id, "size_bytes": LARGE_BLOB,
212 + "first_part": first_part, "count": count,
213 + })
214 + .to_string(),
215 + )
216 + .await;
217 + assert!(
218 + resp.status.is_client_error(),
219 + "first_part={first_part} count={count} must be refused: {}",
220 + resp.text
221 + );
222 + }
223 + }
224 +
225 + #[tokio::test]
226 + async fn multipart_complete_then_confirm_records_the_blob() {
227 + // The end-to-end handoff. Confirm is untouched by multipart: it reads the
228 + // real object size and charges that, not anything the session declared.
229 + let (mut h, blobs) = harness_with_blobs().await;
230 + let user_id = h.signup("mp_flow", "mp_flow@example.com", "Password1!").await;
231 + let (app_id, _key) = create_internal_app(&h.db, user_id).await;
232 + seed_subscription(&h.db, user_id, app_id, "active", SYNCKIT_MAX_MULTIPART_BLOB_SIZE_BYTES).await;
233 + auth_as(&mut h, user_id, app_id, "user-key");
234 +
235 + let hash = fake_hash(0x16);
236 + let start: Value = h
237 + .client
238 + .post_json(
239 + "/api/sync/blobs/multipart/start",
240 + &json!({ "hash": hash, "size_bytes": LARGE_BLOB }).to_string(),
241 + )
242 + .await
243 + .json();
244 + let upload_id = start["upload_id"].as_str().unwrap().to_string();
245 + assert_eq!(blobs.open_multipart_count(), 1, "start must open exactly one session");
246 +
247 + // The client PUTs each part to its presigned URL; the in-memory backend
248 + // cannot receive them, so stand in for the assembled object here. The
249 + // stored object is deliberately a different size than the declared one —
250 + // confirm must charge what S3 actually holds.
251 + let actual_size = 4096;
252 + blobs.put(&format!("{app_id}/{user_id}/{hash}"), vec![7u8; actual_size]);
253 +
254 + let complete = h
255 + .client
256 + .post_json(
257 + "/api/sync/blobs/multipart/complete",
258 + &json!({
259 + "hash": hash, "upload_id": upload_id,
260 + "parts": [{"part_number": 1, "etag": "\"etag-1\""}],
261 + })
262 + .to_string(),
263 + )
264 + .await;
265 + assert_eq!(complete.status, 204, "complete failed: {}", complete.text);
266 + assert_eq!(blobs.open_multipart_count(), 0, "complete must close the session");
267 +
268 + let confirm = h
269 + .client
270 + .post_json("/api/sync/blobs/confirm", &json!({ "hash": hash }).to_string())
271 + .await;
272 + assert_eq!(confirm.status, 204, "confirm after multipart failed: {}", confirm.text);
273 +
274 + let stored: i64 = sqlx::query_scalar(
275 + "SELECT size_bytes FROM sync_blobs WHERE app_id = $1 AND user_id = $2 AND hash = $3",
276 + )
277 + .bind(app_id)
278 + .bind(user_id)
279 + .bind(&hash)
280 + .fetch_one(&h.db)
281 + .await
282 + .unwrap();
283 + assert_eq!(stored as usize, actual_size, "confirm must record the authoritative object size");
284 +
285 + // Confirm consumes the session's pending-upload row, so the orphan reaper
286 + // has nothing left to chase.
287 + let pending: i64 = sqlx::query_scalar(
288 + "SELECT COUNT(*) FROM pending_uploads WHERE user_id = $1 AND bucket = 'synckit'",
289 + )
290 + .bind(user_id)
291 + .fetch_one(&h.db)
292 + .await
293 + .unwrap();
294 + assert_eq!(pending, 0, "confirm must clear the pending-upload row");
295 + }
296 +
297 + #[tokio::test]
298 + async fn multipart_abort_releases_the_session_and_its_tracking_row() {
299 + // An abandoned session leaves no object at all, only parts S3 bills for.
300 + // A client that cancels cleanly should release them immediately rather than
301 + // waiting on the reaper.
302 + let (mut h, blobs) = harness_with_blobs().await;
303 + let user_id = h.signup("mp_abort", "mp_abort@example.com", "Password1!").await;
304 + let (app_id, _key) = create_internal_app(&h.db, user_id).await;
305 + seed_subscription(&h.db, user_id, app_id, "active", SYNCKIT_MAX_MULTIPART_BLOB_SIZE_BYTES).await;
306 + auth_as(&mut h, user_id, app_id, "user-key");
307 +
308 + let hash = fake_hash(0x17);
309 + let start: Value = h
310 + .client
311 + .post_json(
312 + "/api/sync/blobs/multipart/start",
313 + &json!({ "hash": hash, "size_bytes": LARGE_BLOB }).to_string(),
314 + )
315 + .await
316 + .json();
317 + let upload_id = start["upload_id"].as_str().unwrap().to_string();
318 + assert_eq!(blobs.open_multipart_count(), 1);
319 +
320 + let abort = h
321 + .client
322 + .post_json(
323 + "/api/sync/blobs/multipart/abort",
324 + &json!({ "hash": hash, "upload_id": upload_id }).to_string(),
325 + )
326 + .await;
327 + assert_eq!(abort.status, 204, "abort failed: {}", abort.text);
328 + assert_eq!(blobs.open_multipart_count(), 0, "abort must release the parts");
329 +
330 + let pending: i64 = sqlx::query_scalar(
331 + "SELECT COUNT(*) FROM pending_uploads WHERE user_id = $1 AND bucket = 'synckit'",
332 + )
333 + .bind(user_id)
334 + .fetch_one(&h.db)
335 + .await
336 + .unwrap();
337 + assert_eq!(pending, 0, "abort must clear the tracking row it opened");
338 + }
M server/tests/workflows/synckit_paid_sync.rs +5 -5
@@ -15,7 +15,7 @@ use crate::harness::{client::TestResponse, storage::InMemoryStorage, stripe, Bui
15 15
16 16 const GIB: i64 = 1024 * 1024 * 1024;
17 17
18 - async fn harness_with_blobs() -> (TestHarness, Arc<InMemoryStorage>) {
18 + pub(crate) async fn harness_with_blobs() -> (TestHarness, Arc<InMemoryStorage>) {
19 19 let synckit_mem = Arc::new(InMemoryStorage::new());
20 20 let mock_stripe = Arc::new(stripe::MockPaymentProvider::new());
21 21 let mut h = TestHarness::build(BuildOptions {
@@ -29,7 +29,7 @@ async fn harness_with_blobs() -> (TestHarness, Arc<InMemoryStorage>) {
29 29 }
30 30
31 31 /// Create an active first-party (internal) sync app.
32 - async fn create_internal_app(pool: &PgPool, user_id: UserId) -> (SyncAppId, String) {
32 + pub(crate) async fn create_internal_app(pool: &PgPool, user_id: UserId) -> (SyncAppId, String) {
33 33 let api_key = "test-api-key-paid-sync";
34 34 let key_hash = crate::harness::hash_api_key(api_key);
35 35 let key_prefix = &api_key[..8];
@@ -48,7 +48,7 @@ async fn create_internal_app(pool: &PgPool, user_id: UserId) -> (SyncAppId, Stri
48 48 }
49 49
50 50 /// Seed an end-user subscription row with the given status and cap.
51 - async fn seed_subscription(pool: &PgPool, user_id: UserId, app_id: SyncAppId, status: &str, limit_bytes: i64) {
51 + pub(crate) async fn seed_subscription(pool: &PgPool, user_id: UserId, app_id: SyncAppId, status: &str, limit_bytes: i64) {
52 52 sqlx::query(
53 53 "INSERT INTO app_sync_subscriptions
54 54 (user_id, app_id, stripe_subscription_id, stripe_customer_id, tier, status, storage_limit_bytes)
@@ -64,7 +64,7 @@ async fn seed_subscription(pool: &PgPool, user_id: UserId, app_id: SyncAppId, st
64 64 .expect("seed subscription");
65 65 }
66 66
67 - fn auth_as(h: &mut TestHarness, user_id: UserId, app_id: SyncAppId, key: &str) {
67 + pub(crate) fn auth_as(h: &mut TestHarness, user_id: UserId, app_id: SyncAppId, key: &str) {
68 68 let token = makenotwork::synckit_auth::create_sync_token(
69 69 "test-synckit-jwt-secret",
70 70 user_id,
@@ -75,7 +75,7 @@ fn auth_as(h: &mut TestHarness, user_id: UserId, app_id: SyncAppId, key: &str) {
75 75 h.client.set_bearer_token(&token);
76 76 }
77 77
78 - fn fake_hash(seed: u8) -> String {
78 + pub(crate) fn fake_hash(seed: u8) -> String {
79 79 let mut s = String::with_capacity(64);
80 80 for _ in 0..32 {
81 81 s.push_str(&format!("{seed:02x}"));
M shared/synckit-client/src/crypto.rs +109 -13
@@ -496,6 +496,15 @@ fn blob_chunk_aad(hash: &str, index: u32, chunk_count: u32) -> Vec<u8> {
496 496 aad
497 497 }
498 498
499 + /// Number of v3 chunks a `total_len`-byte plaintext seals into, at this
500 + /// client's [`BLOB_CHUNK_SIZE`].
501 + ///
502 + /// Pure arithmetic over the plaintext length, which is what lets a streaming
503 + /// uploader lay out every part boundary before reading a byte of the file.
504 + pub fn blob_chunk_count_for(total_len: usize) -> u32 {
505 + blob_chunk_count(total_len, BLOB_CHUNK_SIZE)
506 + }
507 +
499 508 /// Number of chunks for a `total_len` plaintext at `chunk_size` (min 1).
500 509 fn blob_chunk_count(total_len: usize, chunk_size: usize) -> u32 {
501 510 if total_len == 0 {
@@ -549,6 +558,47 @@ pub fn chunked_blob_overhead(plaintext_len: usize) -> usize {
549 558 WIRE_V3_TAG_BYTES.len() + BLOB_V3_HEADER_LEN + chunk_count * ENCRYPTION_OVERHEAD
550 559 }
551 560
561 + /// Exact v3 ciphertext length for a `plaintext_len`-byte blob.
562 + ///
563 + /// Known from the plaintext length alone, so a multipart uploader can declare
564 + /// the object size (and therefore sign an exact `Content-Length` per part)
565 + /// before it has sealed anything.
566 + pub fn blob_encrypted_len(plaintext_len: usize) -> usize {
567 + plaintext_len + chunked_blob_overhead(plaintext_len)
568 + }
569 +
570 + /// The v3 tag + header that precedes the sealed-chunk stream for a
571 + /// `total_len`-byte plaintext.
572 + ///
573 + /// Split out of [`encrypt_blob_chunked`] so a streaming uploader can emit the
574 + /// header first and then seal chunks one at a time; the two paths produce
575 + /// byte-identical output.
576 + pub fn blob_header_bytes(total_len: usize) -> Vec<u8> {
577 + let mut out = Vec::with_capacity(WIRE_V3_TAG_BYTES.len() + BLOB_V3_HEADER_LEN);
578 + out.extend_from_slice(WIRE_V3_TAG_BYTES);
579 + out.push(3);
580 + out.extend_from_slice(&(BLOB_CHUNK_SIZE as u32).to_le_bytes());
581 + out.extend_from_slice(&(total_len as u64).to_le_bytes());
582 + out
583 + }
584 +
585 + /// Seal one blob chunk, binding `(hash, index, chunk_count)` as AAD.
586 + ///
587 + /// The encrypt-side counterpart to [`decrypt_blob_chunk`]. `chunk` must be the
588 + /// `index`-th [`BLOB_CHUNK_SIZE`] slice of the plaintext (the last may be
589 + /// shorter), and `chunk_count` must be [`blob_chunk_count_for`] of the whole
590 + /// plaintext — the AAD binds both, so a streaming caller that miscounts
591 + /// produces a blob that fails to open rather than one that opens wrong.
592 + pub fn seal_blob_chunk(
593 + chunk: &[u8],
594 + master_key: &[u8; KEY_SIZE],
595 + hash: &str,
596 + index: u32,
597 + chunk_count: u32,
598 + ) -> Result<Vec<u8>> {
599 + seal(chunk, master_key, &blob_chunk_aad(hash, index, chunk_count))
600 + }
601 +
552 602 /// Parse the v3 header, returning it plus the number of bytes consumed (tag +
553 603 /// header). The remaining input is the sealed-chunk stream.
554 604 pub fn parse_blob_header(data: &[u8]) -> Result<(BlobHeader, usize)> {
@@ -592,24 +642,16 @@ pub fn encrypt_blob_chunked(
592 642 hash: &str,
593 643 ) -> Result<Vec<u8>> {
594 644 let total_len = plaintext.len();
595 - let chunk_count = blob_chunk_count(total_len, BLOB_CHUNK_SIZE);
596 - let mut out = Vec::with_capacity(
597 - WIRE_V3_TAG_BYTES.len()
598 - + BLOB_V3_HEADER_LEN
599 - + total_len
600 - + chunk_count as usize * ENCRYPTION_OVERHEAD,
601 - );
602 - out.extend_from_slice(WIRE_V3_TAG_BYTES);
603 - out.push(3);
604 - out.extend_from_slice(&(BLOB_CHUNK_SIZE as u32).to_le_bytes());
605 - out.extend_from_slice(&(total_len as u64).to_le_bytes());
645 + let chunk_count = blob_chunk_count_for(total_len);
646 + let mut out = Vec::with_capacity(blob_encrypted_len(total_len));
647 + out.extend_from_slice(&blob_header_bytes(total_len));
606 648
607 649 if total_len == 0 {
608 - out.extend_from_slice(&seal(&[], master_key, &blob_chunk_aad(hash, 0, 1))?);
650 + out.extend_from_slice(&seal_blob_chunk(&[], master_key, hash, 0, 1)?);
609 651 return Ok(out);
610 652 }
611 653 for (i, chunk) in plaintext.chunks(BLOB_CHUNK_SIZE).enumerate() {
612 - out.extend_from_slice(&seal(chunk, master_key, &blob_chunk_aad(hash, i as u32, chunk_count))?);
654 + out.extend_from_slice(&seal_blob_chunk(chunk, master_key, hash, i as u32, chunk_count)?);
613 655 }
614 656 Ok(out)
615 657 }
@@ -1575,6 +1617,60 @@ mod tests {
1575 1617 }
1576 1618
1577 1619 #[test]
1620 + fn streaming_seal_matches_whole_buffer_encrypt() {
1621 + let key = generate_master_key();
1622 + let hash = "sha256-stream";
1623 + // Two full chunks plus a partial, and the empty blob, which the
1624 + // streaming path has to special-case the same way (one empty chunk).
1625 + for pt in [
1626 + vec![],
1627 + b"one small chunk".to_vec(),
1628 + (0..(BLOB_CHUNK_SIZE * 2 + 7)).map(|i| i as u8).collect(),
1629 + ] {
1630 + let whole = encrypt_blob_chunked(&pt, &key, hash).unwrap();
1631 +
1632 + // What a streaming uploader does: header first, then seal each
1633 + // chunk as it reads it. Nonces are random, so the bytes differ —
1634 + // the contract is the LENGTH (which every part boundary is signed
1635 + // against) and that the result opens to the same plaintext.
1636 + let chunk_count = blob_chunk_count_for(pt.len());
1637 + let mut streamed = blob_header_bytes(pt.len());
1638 + if pt.is_empty() {
1639 + streamed.extend_from_slice(&seal_blob_chunk(&[], &key, hash, 0, 1).unwrap());
1640 + } else {
1641 + for (i, chunk) in pt.chunks(BLOB_CHUNK_SIZE).enumerate() {
1642 + streamed.extend_from_slice(
1643 + &seal_blob_chunk(chunk, &key, hash, i as u32, chunk_count).unwrap(),
1644 + );
1645 + }
1646 + }
1647 +
1648 + assert_eq!(streamed.len(), whole.len());
1649 + assert_eq!(streamed.len(), blob_encrypted_len(pt.len()));
1650 + assert_eq!(decrypt_blob_chunked(&streamed, &key, hash).unwrap(), pt);
1651 + }
1652 + }
1653 +
1654 + #[test]
1655 + fn blob_encrypted_len_predicts_the_wire_size() {
1656 + let key = generate_master_key();
1657 + // Boundary sizes: the prediction is what a multipart start declares, so
1658 + // being off by one byte anywhere breaks a signed Content-Length.
1659 + for len in [
1660 + 0,
1661 + 1,
1662 + BLOB_CHUNK_SIZE - 1,
1663 + BLOB_CHUNK_SIZE,
1664 + BLOB_CHUNK_SIZE + 1,
1665 + BLOB_CHUNK_SIZE * 3,
1666 + ] {
1667 + let pt = vec![7u8; len];
1668 + let wire = encrypt_blob_chunked(&pt, &key, "h").unwrap();
1669 + assert_eq!(blob_encrypted_len(len), wire.len(), "len {len}");
1670 + }
1671 + }
1672 +
1673 + #[test]
1578 1674 fn chunked_blob_wrong_hash_fails_closed() {
1579 1675 let key = generate_master_key();
1580 1676 let pt = b"bound to its address".to_vec();