Skip to main content

max / makenotwork

synckit: chunked AEAD blob format (sk3) with streaming, bounded-RAM download Deep A+ pass, transport axis (D8). A blob was sealed as one AEAD blob and the whole ciphertext was buffered in RAM before decrypt (~2x peak). New v3 format splits the plaintext into 1 MiB chunks, each sealed and AAD-bound to (content_hash, chunk_index, chunk_count) so reorder/truncation/duplication/ substitution all fail closed. blob_download streams the body and decrypts chunk-by-chunk (peak RAM ~= plaintext + one chunk, not 2x); legacy sk2:/untagged blobs still decrypt whole-buffer (no flag-day). Adds the reqwest "stream" feature. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-22 23:09 UTC
Commit: 16c40bc36e61c80845bcab5940e2a79188cde656
Parent: adf3718
5 files changed, +329 insertions, -19 deletions
M shared/synckit-client/Cargo.lock +16
@@ -1230,6 +1230,7 @@ dependencies = [
1230 1230 "bytes",
1231 1231 "encoding_rs",
1232 1232 "futures-core",
1233 + "futures-util",
1233 1234 "h2",
1234 1235 "http",
1235 1236 "http-body",
@@ -1251,12 +1252,14 @@ dependencies = [
1251 1252 "sync_wrapper",
1252 1253 "tokio",
1253 1254 "tokio-native-tls",
1255 + "tokio-util",
1254 1256 "tower",
1255 1257 "tower-http",
1256 1258 "tower-service",
1257 1259 "url",
1258 1260 "wasm-bindgen",
1259 1261 "wasm-bindgen-futures",
1262 + "wasm-streams",
1260 1263 "web-sys",
1261 1264 ]
1262 1265
@@ -1991,6 +1994,19 @@ dependencies = [
1991 1994 ]
1992 1995
1993 1996 [[package]]
1997 + name = "wasm-streams"
1998 + version = "0.4.2"
1999 + source = "registry+https://github.com/rust-lang/crates.io-index"
2000 + checksum = "15053d8d85c7eccdbefef60f06769760a563c7f0a9d6902a13d35c7800b0ad65"
2001 + dependencies = [
2002 + "futures-util",
2003 + "js-sys",
2004 + "wasm-bindgen",
2005 + "wasm-bindgen-futures",
2006 + "web-sys",
2007 + ]
2008 +
2009 + [[package]]
1994 2010 name = "wasmparser"
1995 2011 version = "0.244.0"
1996 2012 source = "registry+https://github.com/rust-lang/crates.io-index"
M shared/synckit-client/Cargo.toml +1 -1
@@ -26,7 +26,7 @@ zeroize = "1"
26 26 sha2 = "0.10"
27 27
28 28 # HTTP
29 - reqwest = { version = "0.12", features = ["json", "native-tls"] }
29 + reqwest = { version = "0.12", features = ["json", "native-tls", "stream"] }
30 30 bytes = "1"
31 31 tokio = { version = "1", features = ["rt-multi-thread", "macros", "time"] }
32 32 tokio-stream = "0.1"
M shared/synckit-client/src/client/blob.rs +76 -13
@@ -1,5 +1,6 @@
1 - use bytes::Bytes;
1 + use bytes::{Buf, Bytes, BytesMut};
2 2 use sha2::{Digest, Sha256};
3 + use tokio_stream::StreamExt;
3 4 use tracing::instrument;
4 5
5 6 use crate::{
@@ -59,8 +60,9 @@ impl SyncKitClient {
59 60 #[instrument(skip(self, presigned_url, data))]
60 61 pub async fn blob_upload(&self, hash: &str, presigned_url: &str, data: Vec<u8>) -> Result<()> {
61 62 let master_key = self.require_master_key()?;
62 - let encrypted =
63 - Bytes::from(crypto::encrypt_bytes_aad(&data, &master_key, &crypto::AeadContext::blob(hash))?);
63 + // v3 chunked format: each chunk sealed and bound to (hash, index, count),
64 + // so the reader verifies and decrypts one chunk at a time.
65 + let encrypted = Bytes::from(crypto::encrypt_blob_chunked(&data, &master_key, hash)?);
64 66
65 67 self.retry_request(Idempotency::Safe, || {
66 68 let req = self
@@ -136,8 +138,11 @@ impl SyncKitClient {
136 138 /// re-hash closes the substitution/rollback gap. Returns
137 139 /// [`SyncKitError::IntegrityFailed`] on mismatch.
138 140 ///
139 - /// The blob is held in memory (the caller writes it to disk); the decrypted
140 - /// size is capped at [`MAX_BLOB_BYTES`].
141 + /// The decrypted blob is returned in memory (the caller writes it to disk);
142 + /// the size is capped at [`MAX_BLOB_BYTES`]. A v3 chunked blob is decrypted
143 + /// chunk-by-chunk off the HTTP body, so peak memory is the plaintext plus one
144 + /// in-flight chunk rather than ciphertext *and* plaintext at once. Legacy
145 + /// `sk2:`/untagged blobs fall back to a whole-buffer decrypt (no flag-day).
141 146 #[instrument(skip(self, presigned_url))]
142 147 pub async fn blob_download(&self, expected_hash: &str, presigned_url: &str) -> Result<Vec<u8>> {
143 148 let resp = self
@@ -147,15 +152,73 @@ impl SyncKitClient {
147 152 })
148 153 .await?;
149 154
150 - // Owned `Bytes`; decrypt borrows it directly (no redundant copy).
151 - let encrypted = resp.bytes().await?;
152 - if encrypted.len() > MAX_BLOB_BYTES {
153 - return Err(SyncKitError::Internal(format!(
154 - "blob exceeds {MAX_BLOB_BYTES}-byte limit",
155 - )));
156 - }
157 155 let master_key = self.require_master_key()?;
158 - let plaintext = crypto::decrypt_bytes_aad(&encrypted, &master_key, &crypto::AeadContext::blob(expected_hash))?;
156 + let mut stream = resp.bytes_stream();
157 + let mut buf = BytesMut::new();
158 + let mut header: Option<crypto::BlobHeader> = None;
159 + let mut format_known = false;
160 + let mut chunked = false;
161 + let mut next_chunk: u32 = 0;
162 + let mut plaintext: Vec<u8> = Vec::new();
163 + let mut total_in: usize = 0;
164 +
165 + while let Some(part) = stream.next().await {
166 + let part = part.map_err(SyncKitError::Http)?;
167 + total_in = total_in.saturating_add(part.len());
168 + if total_in > MAX_BLOB_BYTES {
169 + return Err(SyncKitError::Internal(format!(
170 + "blob exceeds {MAX_BLOB_BYTES}-byte limit",
171 + )));
172 + }
173 + buf.extend_from_slice(&part);
174 +
175 + // Pick the format once the 4-byte tag is in hand.
176 + if !format_known && buf.len() >= 4 {
177 + chunked = crypto::is_chunked_blob(&buf);
178 + format_known = true;
179 + }
180 + if !chunked {
181 + continue; // legacy: accumulate, decrypt whole-buffer at the end
182 + }
183 + // Parse the v3 header once available, consuming tag + header bytes.
184 + if header.is_none() {
185 + if buf.len() < 4 + 13 {
186 + continue;
187 + }
188 + let (h, consumed) = crypto::parse_blob_header(&buf)?;
189 + buf.advance(consumed);
190 + header = Some(h);
191 + }
192 + // Peel and decrypt every complete chunk currently buffered.
193 + let h = header.unwrap();
194 + while next_chunk < h.chunk_count {
195 + let len = h.sealed_chunk_len(next_chunk);
196 + if buf.len() < len {
197 + break;
198 + }
199 + let sealed = buf.split_to(len);
200 + let chunk = crypto::decrypt_blob_chunk(
201 + &sealed,
202 + &master_key,
203 + expected_hash,
204 + next_chunk,
205 + h.chunk_count,
206 + )?;
207 + plaintext.extend_from_slice(&chunk);
208 + next_chunk += 1;
209 + }
210 + }
211 +
212 + if chunked {
213 + let h = header.ok_or_else(|| SyncKitError::Crypto("v3 blob ended before its header".into()))?;
214 + if next_chunk != h.chunk_count || buf.has_remaining() {
215 + return Err(SyncKitError::Crypto("v3 blob ended mid-chunk or had trailing bytes".into()));
216 + }
217 + } else {
218 + // Legacy (sk2: / untagged): decrypt the accumulated body in one shot.
219 + plaintext =
220 + crypto::decrypt_bytes_aad(&buf, &master_key, &crypto::AeadContext::blob(expected_hash))?;
221 + }
159 222
160 223 let actual = format!("{:x}", Sha256::digest(&plaintext));
161 224 if actual != expected_hash {
M shared/synckit-client/src/crypto.rs +231
@@ -433,6 +433,173 @@ pub fn decrypt_bytes_aad(
433 433 /// Encryption overhead in bytes (24-byte nonce + 16-byte Poly1305 tag).
434 434 pub const ENCRYPTION_OVERHEAD: usize = NONCE_SIZE + 16;
435 435
436 + // ── Chunked blob format (v3) ──
437 + //
438 + // A blob is split into fixed-size plaintext chunks, each sealed independently
439 + // with AAD binding (content_hash, chunk_index, chunk_count). This lets a reader
440 + // decrypt and verify one chunk at a time — bounding peak memory to a single
441 + // chunk instead of the whole blob — while making chunk reorder, truncation,
442 + // duplication, and cross-blob substitution all fail closed at the AEAD layer.
443 + //
444 + // Layout: `sk3:` | version(u8=3) | chunk_size(u32 LE) | total_len(u64 LE) | sealed_chunk*
445 + // where each `sealed_chunk` is `nonce[24] || ciphertext || tag[16]`. A zero-length
446 + // blob is encoded as a single empty sealed chunk so it is still authenticated.
447 +
448 + /// Wire tag for the v3 chunked blob format.
449 + const WIRE_V3_TAG_BYTES: &[u8] = b"sk3:";
450 + /// Fixed header length after the tag: version(1) + chunk_size(4) + total_len(8).
451 + const BLOB_V3_HEADER_LEN: usize = 1 + 4 + 8;
452 + /// Plaintext bytes per blob chunk (1 MiB). Bounds per-chunk working memory.
453 + pub const BLOB_CHUNK_SIZE: usize = 1024 * 1024;
454 +
455 + /// AAD binding a blob chunk to its content hash and position. `chunk_count` is
456 + /// included so dropping the final chunk (truncation) changes every chunk's AAD.
457 + fn blob_chunk_aad(hash: &str, index: u32, chunk_count: u32) -> Vec<u8> {
458 + let mut aad = Vec::with_capacity(hash.len() + 1 + 8);
459 + aad.extend_from_slice(hash.as_bytes());
460 + aad.push(0x1f);
461 + aad.extend_from_slice(&index.to_le_bytes());
462 + aad.extend_from_slice(&chunk_count.to_le_bytes());
463 + aad
464 + }
465 +
466 + /// Number of chunks for a `total_len` plaintext at `chunk_size` (min 1).
467 + fn blob_chunk_count(total_len: usize, chunk_size: usize) -> u32 {
468 + if total_len == 0 {
469 + 1
470 + } else {
471 + total_len.div_ceil(chunk_size) as u32
472 + }
473 + }
474 +
475 + /// Parsed v3 blob header.
476 + #[derive(Clone, Copy, Debug)]
477 + pub struct BlobHeader {
478 + /// Plaintext bytes per chunk (last chunk may be shorter).
479 + pub chunk_size: usize,
480 + /// Total plaintext length.
481 + pub total_len: usize,
482 + /// Number of sealed chunks.
483 + pub chunk_count: u32,
484 + }
485 +
486 + impl BlobHeader {
487 + /// Sealed length of chunk `index` (`nonce + plaintext_slice + tag`).
488 + pub fn sealed_chunk_len(&self, index: u32) -> usize {
489 + let plain = if self.total_len == 0 {
490 + 0
491 + } else if index == self.chunk_count - 1 {
492 + self.total_len - self.chunk_size * index as usize
493 + } else {
494 + self.chunk_size
495 + };
496 + ENCRYPTION_OVERHEAD + plain
497 + }
498 + }
499 +
500 + /// Whether `data` begins with the v3 chunked-blob tag.
501 + pub fn is_chunked_blob(data: &[u8]) -> bool {
502 + data.starts_with(WIRE_V3_TAG_BYTES)
503 + }
504 +
505 + /// Total wire overhead [`encrypt_blob_chunked`] adds for a `plaintext_len`-byte
506 + /// blob: the tag + header, plus one nonce + tag per chunk.
507 + pub fn chunked_blob_overhead(plaintext_len: usize) -> usize {
508 + let chunk_count = blob_chunk_count(plaintext_len, BLOB_CHUNK_SIZE) as usize;
509 + WIRE_V3_TAG_BYTES.len() + BLOB_V3_HEADER_LEN + chunk_count * ENCRYPTION_OVERHEAD
510 + }
511 +
512 + /// Parse the v3 header, returning it plus the number of bytes consumed (tag +
513 + /// header). The remaining input is the sealed-chunk stream.
514 + pub fn parse_blob_header(data: &[u8]) -> Result<(BlobHeader, usize)> {
515 + let body = data
516 + .strip_prefix(WIRE_V3_TAG_BYTES)
517 + .ok_or_else(|| SyncKitError::Crypto("not a v3 chunked blob".into()))?;
518 + if body.len() < BLOB_V3_HEADER_LEN {
519 + return Err(SyncKitError::Crypto("v3 blob header truncated".into()));
520 + }
521 + if body[0] != 3 {
522 + return Err(SyncKitError::Crypto(format!(
523 + "unknown blob format version {}; this client is too old to read it",
524 + body[0]
525 + )));
526 + }
527 + let chunk_size = u32::from_le_bytes(body[1..5].try_into().unwrap()) as usize;
528 + let total_len = u64::from_le_bytes(body[5..13].try_into().unwrap()) as usize;
529 + if chunk_size == 0 {
530 + return Err(SyncKitError::Crypto("v3 blob chunk_size is zero".into()));
531 + }
532 + let chunk_count = blob_chunk_count(total_len, chunk_size);
533 + let consumed = WIRE_V3_TAG_BYTES.len() + BLOB_V3_HEADER_LEN;
534 + Ok((BlobHeader { chunk_size, total_len, chunk_count }, consumed))
535 + }
536 +
537 + /// Decrypt one sealed chunk, binding `(hash, index, chunk_count)`.
538 + pub fn decrypt_blob_chunk(
539 + sealed: &[u8],
540 + master_key: &[u8; KEY_SIZE],
541 + hash: &str,
542 + index: u32,
543 + chunk_count: u32,
544 + ) -> Result<Vec<u8>> {
545 + open(sealed, master_key, &blob_chunk_aad(hash, index, chunk_count))
546 + }
547 +
548 + /// Encrypt a blob into the v3 chunked wire format (see the module note above).
549 + pub fn encrypt_blob_chunked(
550 + plaintext: &[u8],
551 + master_key: &[u8; KEY_SIZE],
552 + hash: &str,
553 + ) -> Result<Vec<u8>> {
554 + let total_len = plaintext.len();
555 + let chunk_count = blob_chunk_count(total_len, BLOB_CHUNK_SIZE);
556 + let mut out = Vec::with_capacity(
557 + WIRE_V3_TAG_BYTES.len()
558 + + BLOB_V3_HEADER_LEN
559 + + total_len
560 + + chunk_count as usize * ENCRYPTION_OVERHEAD,
561 + );
562 + out.extend_from_slice(WIRE_V3_TAG_BYTES);
563 + out.push(3);
564 + out.extend_from_slice(&(BLOB_CHUNK_SIZE as u32).to_le_bytes());
565 + out.extend_from_slice(&(total_len as u64).to_le_bytes());
566 +
567 + if total_len == 0 {
568 + out.extend_from_slice(&seal(&[], master_key, &blob_chunk_aad(hash, 0, 1))?);
569 + return Ok(out);
570 + }
571 + for (i, chunk) in plaintext.chunks(BLOB_CHUNK_SIZE).enumerate() {
572 + out.extend_from_slice(&seal(chunk, master_key, &blob_chunk_aad(hash, i as u32, chunk_count))?);
573 + }
574 + Ok(out)
575 + }
576 +
577 + /// Decrypt a complete v3 chunked blob from one buffer. Used by the non-streaming
578 + /// fallback and tests; the streaming download path parses chunks incrementally
579 + /// via [`parse_blob_header`] + [`decrypt_blob_chunk`].
580 + pub fn decrypt_blob_chunked(
581 + encrypted: &[u8],
582 + master_key: &[u8; KEY_SIZE],
583 + hash: &str,
584 + ) -> Result<Vec<u8>> {
585 + let (header, consumed) = parse_blob_header(encrypted)?;
586 + let mut rest = &encrypted[consumed..];
587 + let mut out = Vec::with_capacity(header.total_len);
588 + for i in 0..header.chunk_count {
589 + let len = header.sealed_chunk_len(i);
590 + if rest.len() < len {
591 + return Err(SyncKitError::Crypto("v3 blob truncated mid-chunk".into()));
592 + }
593 + let (sealed, tail) = rest.split_at(len);
594 + out.extend_from_slice(&decrypt_blob_chunk(sealed, master_key, hash, i, header.chunk_count)?);
595 + rest = tail;
596 + }
597 + if !rest.is_empty() {
598 + return Err(SyncKitError::Crypto("v3 blob has trailing bytes".into()));
599 + }
600 + Ok(out)
601 + }
602 +
436 603 /// Overhead of a v2 (AAD-bound, `sk2:`-tagged) blob: the wire tag plus the
437 604 /// nonce + Poly1305 tag.
438 605 pub const ENCRYPTION_OVERHEAD_V2: usize = WIRE_V2_TAG_BYTES.len() + ENCRYPTION_OVERHEAD;
@@ -1277,4 +1444,68 @@ mod tests {
1277 1444 let pt = decrypt_bytes_aad(&legacy, &key, &AeadContext::blob("sha256-whatever")).unwrap();
1278 1445 assert_eq!(pt, b"old blob");
1279 1446 }
1447 +
1448 + // ── chunked blob format (v3) ──
1449 +
1450 + #[test]
1451 + fn chunked_blob_roundtrips_across_chunk_boundaries() {
1452 + let key = generate_master_key();
1453 + // Spans three chunks (two full + a partial), exercising the boundary math.
1454 + let plaintext: Vec<u8> = (0..(BLOB_CHUNK_SIZE * 2 + 123)).map(|i| i as u8).collect();
1455 + let hash = "sha256-abc";
1456 + let wire = encrypt_blob_chunked(&plaintext, &key, hash).unwrap();
1457 + assert!(is_chunked_blob(&wire), "must carry the sk3: tag");
1458 + let (header, _) = parse_blob_header(&wire).unwrap();
1459 + assert_eq!(header.total_len, plaintext.len());
1460 + assert_eq!(header.chunk_count, 3);
1461 + assert_eq!(decrypt_blob_chunked(&wire, &key, hash).unwrap(), plaintext);
1462 + }
1463 +
1464 + #[test]
1465 + fn chunked_blob_handles_empty_and_single_chunk() {
1466 + let key = generate_master_key();
1467 + for pt in [vec![], b"small blob".to_vec()] {
1468 + let wire = encrypt_blob_chunked(&pt, &key, "h").unwrap();
1469 + assert_eq!(parse_blob_header(&wire).unwrap().0.chunk_count, 1);
1470 + assert_eq!(decrypt_blob_chunked(&wire, &key, "h").unwrap(), pt);
1471 + }
1472 + }
1473 +
1474 + #[test]
1475 + fn chunked_blob_wrong_hash_fails_closed() {
1476 + let key = generate_master_key();
1477 + let pt = b"bound to its address".to_vec();
1478 + let wire = encrypt_blob_chunked(&pt, &key, "hash-a").unwrap();
1479 + // A different content hash changes every chunk's AAD -> open fails.
1480 + assert!(matches!(
1481 + decrypt_blob_chunked(&wire, &key, "hash-b"),
1482 + Err(SyncKitError::DecryptionFailed)
1483 + ));
1484 + }
1485 +
1486 + #[test]
1487 + fn chunked_blob_tamper_and_truncation_fail_closed() {
1488 + let key = generate_master_key();
1489 + let pt: Vec<u8> = (0..(BLOB_CHUNK_SIZE + 50)).map(|i| i as u8).collect();
1490 + let wire = encrypt_blob_chunked(&pt, &key, "h").unwrap();
1491 +
1492 + // Flip a ciphertext byte -> AEAD tag mismatch.
1493 + let mut tampered = wire.clone();
1494 + let last = tampered.len() - 1;
1495 + tampered[last] ^= 0x01;
1496 + assert!(decrypt_blob_chunked(&tampered, &key, "h").is_err());
1497 +
1498 + // Drop the final chunk's bytes -> truncation detected.
1499 + let header = parse_blob_header(&wire).unwrap().0;
1500 + let truncated = &wire[..wire.len() - header.sealed_chunk_len(header.chunk_count - 1)];
1501 + assert!(decrypt_blob_chunked(truncated, &key, "h").is_err());
1502 + }
1503 +
1504 + #[test]
1505 + fn parse_blob_header_rejects_non_v3() {
1506 + let key = generate_master_key();
1507 + let v2 = encrypt_bytes_aad(b"x", &key, &AeadContext::blob("h")).unwrap();
1508 + assert!(parse_blob_header(&v2).is_err());
1509 + assert!(!is_chunked_blob(&v2));
1510 + }
1280 1511 }
M shared/synckit-client/tests/integration.rs +5 -5
@@ -1309,7 +1309,7 @@ async fn blob_upload_zero_byte_data() {
1309 1309 let result = client.blob_upload("sha256-empty", &presigned, vec![]).await;
1310 1310 assert!(result.is_ok(), "Zero-byte blob upload should succeed");
1311 1311
1312 - // Verify the uploaded data has encryption overhead only (v2: + sk2: tag)
1312 + // Verify the uploaded data is the v3 chunked framing over an empty blob.
1313 1313 let requests = server.received_requests().await.unwrap();
1314 1314 let req = requests
1315 1315 .iter()
@@ -1317,8 +1317,8 @@ async fn blob_upload_zero_byte_data() {
1317 1317 .unwrap();
1318 1318 assert_eq!(
1319 1319 req.body.len(),
1320 - synckit_client::crypto::ENCRYPTION_OVERHEAD_V2,
1321 - "Empty plaintext should produce exactly overhead bytes"
1320 + synckit_client::crypto::chunked_blob_overhead(0),
1321 + "Empty plaintext should produce exactly the chunked overhead bytes"
1322 1322 );
1323 1323 }
1324 1324
@@ -2513,8 +2513,8 @@ async fn blob_upload_1mb_with_correct_overhead() {
2513 2513 let upload_req = requests.iter().find(|r| r.url.path() == upload_path).unwrap();
2514 2514 assert_eq!(
2515 2515 upload_req.body.len(),
2516 - plaintext.len() + synckit_client::crypto::ENCRYPTION_OVERHEAD_V2,
2517 - "1MB upload should have exactly ENCRYPTION_OVERHEAD_V2 bytes added"
2516 + plaintext.len() + synckit_client::crypto::chunked_blob_overhead(plaintext.len()),
2517 + "1MB upload should add exactly the v3 chunked overhead"
2518 2518 );
2519 2519 }
2520 2520