feat: sync encryption at rest

2026-11-03 11:37:16 +01:00 · 2026-11-01 14:04:56 +04:00
1 parent 61092ed19a
commit bd4a3baf42
@@ -33,3 +33,11 @@ S3_FORCE_PATH_STYLE=true
 # URL it cannot open: /health and /readyz stay green while every transfer fails.
 # Defaults to S3_ENDPOINT, which is correct when storage is already public.
 # S3_PUBLIC_ENDPOINT=https://storage.example.com
+
+# SYNC_AT_REST_KEY: encrypts every stored object with this server's key, even
+# when E2E encryption is off in Donut Browser. Transfers then go through this
+# server instead of straight to storage. Format id:base64 (32 bytes), e.g.
+# k1:$(openssl rand -base64 32). To rotate, put the new key first:
+# k2:...,k1:... The first key encrypts; every key listed decrypts.
+# Objects stored before stay readable; `donut-sync migrate-at-rest` encrypts them.
+# SYNC_AT_REST_KEY
@@ -256,6 +256,12 @@ version = "0.3.34"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e"

+[[package]]
+name = "futures-io"
+version = "0.3.34"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "53c0fa8157de1303bfffdaa1cc2a673bfffb60102f76b0ef4441659124373fed"
+
 [[package]]
 name = "futures-macro"
 version = "0.3.34"
@@ -267,6 +273,12 @@ dependencies = [
 "syn 3.0.9",
 ]

+[[package]]
+name = "futures-sink"
+version = "0.3.34"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1944426bf7d03f1d14f708785e4b33efd750b36d48a157b836b3efc15ede8e1d"
+
 [[package]]
 name = "futures-task"
 version = "0.3.34"
@@ -280,8 +292,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc"
 dependencies = [
 "futures-core",
+ "futures-io",
 "futures-macro",
+ "futures-sink",
 "futures-task",
+ "memchr",
 "pin-project-lite",
 "slab",
 ]
@@ -763,6 +778,7 @@ dependencies = [
 "base64",
 "bytes",
 "futures-core",
+ "futures-util",
 "http",
 "http-body",
 "http-body-util",
@@ -781,12 +797,14 @@ dependencies = [
 "sync_wrapper",
 "tokio",
 "tokio-rustls",
+ "tokio-util",
 "tower",
 "tower-http",
 "tower-service",
 "url",
 "wasm-bindgen",
 "wasm-bindgen-futures",
+ "wasm-streams",
 "web-sys",
 ]

@@ -1180,6 +1198,19 @@ dependencies = [
 "tokio",
 ]

+[[package]]
+name = "tokio-util"
+version = "0.7.20"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e464cf451ba96ebfc6f9b6542f17ee8b8956e33f1e40d9690624e59d7a7f8a4b"
+dependencies = [
+ "bytes",
+ "futures-core",
+ "futures-sink",
+ "pin-project-lite",
+ "tokio",
+]
+
 [[package]]
 name = "tower"
 version = "0.5.3"
@@ -1386,6 +1417,19 @@ dependencies = [
 "unicode-ident",
 ]

+[[package]]
+name = "wasm-streams"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9d1ec4f6517c9e11ae630e200b2b65d193279042e28edd4a2cda233e46670bbb"
+dependencies = [
+ "futures-util",
+ "js-sys",
+ "wasm-bindgen",
+ "wasm-bindgen-futures",
+ "web-sys",
+]
+
 [[package]]
 name = "web-sys"
 version = "0.3.107"
@@ -30,7 +30,7 @@ hyper-util = { version = "0.1.21", features = ["tokio", "server", "http1", "serv
 md-5 = "0.11.0"
 rand = "0.10.3"
 regex-lite = "0.1.9"
-reqwest = { version = "0.13.5", default-features = false, features = ["json", "rustls-no-provider"] }
+reqwest = { version = "0.13.5", default-features = false, features = ["json", "rustls-no-provider", "stream"] }
 ring = "0.17.14"
 roxmltree = "0.21.2"
 rustls = { version = "0.23.46", default-features = false, features = ["ring", "std", "tls12", "logging"] }
@@ -20,6 +20,10 @@ SYNC_TOKEN=... S3_ENDPOINT=... S3_ACCESS_KEY_ID=... S3_SECRET_ACCESS_KEY=... \
 `.env.example` lists the variables. A `.env` file in the working directory is
 also read.

+With `SYNC_AT_REST_KEY` set, the server encrypts every object it stores, and
+transfers go through it instead of straight to storage. Objects stored before
+stay readable; `donut-sync migrate-at-rest --help` encrypts them in place.
+
 ## Test

 The end-to-end tests run the binary against the S3 server in
@@ -0,0 +1,440 @@
+//! Encryption at rest with the server's own key (`SYNC_AT_REST_KEY`).
+//!
+//! An encrypted object is a header and a run of AES-256-GCM frames:
+//!
+//! ```text
+//! "DSAR" | version (1) | key id length (1) | key id | salt (32)
+//! frame 0 | frame 1 | ... | final frame
+//! ```
+//!
+//! Each frame holds up to `FRAME` plaintext bytes and a 16-byte tag. The
+//! object key is HKDF-SHA256 of the server key and the object's random salt;
+//! a frame's nonce is its index and a final flag, so frames cannot be dropped,
+//! reordered or cut off at a frame boundary. Every frame authenticates the
+//! header. Only the last frame may be short; an empty object is one empty
+//! final frame. The object's metadata carries `MARKER` (`1:<key id>`), so a
+//! HEAD tells an encrypted object from a plaintext one and gives its
+//! plaintext size.
+
+use std::collections::VecDeque;
+use std::sync::Arc;
+
+use base64::Engine;
+use bytes::Bytes;
+use futures_util::StreamExt;
+use ring::aead::{AES_256_GCM, Aad, LessSafeKey, Nonce, UnboundKey};
+use ring::hkdf;
+
+use crate::store::{ByteStream, StoreError};
+
+/// The user-metadata name (`x-amz-meta-donut-at-rest`) of an encrypted object.
+pub const MARKER: &str = "donut-at-rest";
+const MAGIC: &[u8; 4] = b"DSAR";
+const VERSION: u8 = 1;
+/// Plaintext bytes per frame.
+pub const FRAME: usize = 64 * 1024;
+const TAG: usize = 16;
+const SALT: usize = 32;
+const MAX_KEY_ID: usize = 32;
+const OBJECT_KEY_INFO: &[u8] = b"donut-sync at-rest v1 object key";
+const URL_KEY_INFO: &[u8] = b"donut-sync at-rest v1 transfer url";
+
+pub struct Key {
+  id: String,
+  secret: [u8; 32],
+}
+
+impl Key {
+  pub fn id(&self) -> &str {
+    &self.id
+  }
+}
+
+/// The server keys. The first one encrypts; all of them decrypt.
+pub struct Keyring {
+  keys: Vec<Key>,
+}
+
+fn key_id_valid(id: &str) -> bool {
+  (1..=MAX_KEY_ID).contains(&id.len())
+    && id
+      .bytes()
+      .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'-' | b'_' | b'.'))
+}
+
+impl Keyring {
+  /// `id:base64,id:base64`, each key 32 bytes.
+  pub fn parse(value: &str) -> Result<Self, String> {
+    let mut keys: Vec<Key> = Vec::new();
+    for entry in value.split(',').map(str::trim).filter(|e| !e.is_empty()) {
+      let (id, encoded) = entry
+        .split_once(':')
+        .ok_or("SYNC_AT_REST_KEY entries are id:base64, e.g. k1:$(openssl rand -base64 32)")?;
+      if !key_id_valid(id) {
+        return Err(format!(
+          "SYNC_AT_REST_KEY key id '{id}' must be 1-{MAX_KEY_ID} letters, digits, '-', '_' or '.'"
+        ));
+      }
+      if keys.iter().any(|k| k.id == id) {
+        return Err(format!("SYNC_AT_REST_KEY names key id '{id}' twice"));
+      }
+      let unpadded = encoded.trim().trim_end_matches('=');
+      let decoded = base64::engine::general_purpose::STANDARD_NO_PAD
+        .decode(unpadded)
+        .or_else(|_| base64::engine::general_purpose::URL_SAFE_NO_PAD.decode(unpadded))
+        .map_err(|_| format!("SYNC_AT_REST_KEY key '{id}' is not base64"))?;
+      let secret: [u8; 32] = decoded
+        .try_into()
+        .map_err(|_| format!("SYNC_AT_REST_KEY key '{id}' must be 32 bytes"))?;
+      keys.push(Key {
+        id: id.to_string(),
+        secret,
+      });
+    }
+    if keys.is_empty() {
+      return Err("SYNC_AT_REST_KEY holds no key".to_string());
+    }
+    Ok(Self { keys })
+  }
+
+  pub fn key_count(&self) -> usize {
+    self.keys.len()
+  }
+
+  pub fn current(&self) -> &Key {
+    &self.keys[0]
+  }
+
+  fn find(&self, id: &str) -> Option<&Key> {
+    self.keys.iter().find(|k| k.id == id)
+  }
+
+  /// The HMAC keys that sign transfer URLs: the current key's first.
+  pub fn url_keys(&self) -> Vec<[u8; 32]> {
+    self
+      .keys
+      .iter()
+      .map(|key| derive(&key.secret, &[0; SALT], URL_KEY_INFO))
+      .collect()
+  }
+
+  /// The marker value an object sealed now carries.
+  pub fn marker(&self) -> String {
+    format!("{VERSION}:{}", self.current().id)
+  }
+}
+
+struct Len(usize);
+
+impl hkdf::KeyType for Len {
+  fn len(&self) -> usize {
+    self.0
+  }
+}
+
+fn derive(secret: &[u8; 32], salt: &[u8], info: &[u8]) -> [u8; 32] {
+  let prk = hkdf::Salt::new(hkdf::HKDF_SHA256, salt).extract(secret);
+  let mut out = [0u8; 32];
+  prk
+    .expand(&[info], Len(32))
+    .and_then(|okm| okm.fill(&mut out))
+    .expect("32 bytes is a valid HKDF-SHA256 length");
+  out
+}
+
+fn object_key(secret: &[u8; 32], salt: &[u8]) -> LessSafeKey {
+  let key = derive(secret, salt, OBJECT_KEY_INFO);
+  LessSafeKey::new(UnboundKey::new(&AES_256_GCM, &key).expect("32 bytes is an AES-256 key"))
+}
+
+fn nonce(index: u64, last: bool) -> Nonce {
+  let mut bytes = [0u8; 12];
+  bytes[3..11].copy_from_slice(&index.to_be_bytes());
+  bytes[11] = u8::from(last);
+  Nonce::assume_unique_for_key(bytes)
+}
+
+fn header_len(key_id: &str) -> u64 {
+  (MAGIC.len() + 2 + key_id.len() + SALT) as u64
+}
+
+fn frames(plain: u64) -> u64 {
+  plain.div_ceil(FRAME as u64).max(1)
+}
+
+/// The stored size of `plain` bytes sealed under `key_id`.
+pub fn sealed_len(key_id: &str, plain: u64) -> u64 {
+  header_len(key_id) + plain + frames(plain) * TAG as u64
+}
+
+/// The plaintext size of a sealed object of `sealed` bytes, or `None` when no
+/// sealed object has that size.
+pub fn plain_len(key_id: &str, sealed: u64) -> Option<u64> {
+  let body = sealed.checked_sub(header_len(key_id))?;
+  let full = (FRAME + TAG) as u64;
+  let (count, rest) = (body / full, body % full);
+  match rest {
+    0 if count > 0 => Some(count * FRAME as u64),
+    rest if rest >= TAG as u64 => Some(count * FRAME as u64 + rest - TAG as u64),
+    _ => None,
+  }
+}
+
+/// The key id a marker value names, for this format version.
+pub fn marker_key_id(marker: &str) -> Option<&str> {
+  marker
+    .strip_prefix(&format!("{VERSION}:"))
+    .filter(|id| key_id_valid(id))
+}
+
+fn corrupt(message: &str) -> StoreError {
+  StoreError::Corrupt(message.to_string())
+}
+
+/// Encrypts a byte stream, frame by frame.
+pub struct Sealer {
+  key: LessSafeKey,
+  header: Vec<u8>,
+  index: u64,
+  pending: Vec<u8>,
+}
+
+impl Sealer {
+  /// A sealer and the header that starts its output.
+  pub fn new(keyring: &Keyring) -> (Self, Bytes) {
+    let current = keyring.current();
+    let mut salt = [0u8; SALT];
+    rand::fill(&mut salt);
+    let mut header = Vec::with_capacity(header_len(&current.id) as usize);
+    header.extend_from_slice(MAGIC);
+    header.push(VERSION);
+    header.push(current.id.len() as u8);
+    header.extend_from_slice(current.id.as_bytes());
+    header.extend_from_slice(&salt);
+    let sealer = Self {
+      key: object_key(&current.secret, &salt),
+      header: header.clone(),
+      index: 0,
+      pending: Vec::new(),
+    };
+    (sealer, Bytes::from(header))
+  }
+
+  fn seal(&mut self, mut frame: Vec<u8>, last: bool) -> Bytes {
+    self
+      .key
+      .seal_in_place_append_tag(nonce(self.index, last), Aad::from(&self.header), &mut frame)
+      .expect("a frame is far below the AES-GCM limit");
+    self.index += 1;
+    Bytes::from(frame)
+  }
+
+  /// Seals every full frame that is known not to be the last.
+  pub fn push(&mut self, data: &[u8], out: &mut VecDeque<Bytes>) {
+    self.pending.extend_from_slice(data);
+    while self.pending.len() > FRAME {
+      let rest = self.pending.split_off(FRAME);
+      let frame = std::mem::replace(&mut self.pending, rest);
+      out.push_back(self.seal(frame, false));
+    }
+  }
+
+  pub fn finish(mut self) -> Bytes {
+    let frame = std::mem::take(&mut self.pending);
+    self.seal(frame, true)
+  }
+}
+
+/// Decrypts a sealed byte stream, frame by frame.
+pub struct Opener {
+  keyring: Arc<Keyring>,
+  key: Option<(LessSafeKey, Vec<u8>)>,
+  index: u64,
+  pending: Vec<u8>,
+}
+
+impl Opener {
+  pub fn new(keyring: Arc<Keyring>) -> Self {
+    Self {
+      keyring,
+      key: None,
+      index: 0,
+      pending: Vec::new(),
+    }
+  }
+
+  fn read_header(&mut self) -> Result<bool, StoreError> {
+    if self.pending.len() < MAGIC.len() + 2 {
+      return Ok(false);
+    }
+    if &self.pending[..MAGIC.len()] != MAGIC {
+      return Err(corrupt("sealed object has no header"));
+    }
+    if self.pending[MAGIC.len()] != VERSION {
+      return Err(corrupt("sealed object has an unknown version"));
+    }
+    let id_len = self.pending[MAGIC.len() + 1] as usize;
+    let total = MAGIC.len() + 2 + id_len + SALT;
+    if self.pending.len() < total {
+      return Ok(false);
+    }
+    let id = std::str::from_utf8(&self.pending[MAGIC.len() + 2..MAGIC.len() + 2 + id_len])
+      .map_err(|_| corrupt("sealed object has an invalid key id"))?;
+    let key = self
+      .keyring
+      .find(id)
+      .ok_or_else(|| StoreError::Corrupt(format!("sealed object needs unknown key id '{id}'")))?;
+    let key = object_key(&key.secret, &self.pending[total - SALT..total]);
+    let rest = self.pending.split_off(total);
+    let header = std::mem::replace(&mut self.pending, rest);
+    self.key = Some((key, header));
+    Ok(true)
+  }
+
+  fn open(&mut self, mut frame: Vec<u8>, last: bool) -> Result<Bytes, StoreError> {
+    let (key, header) = self.key.as_ref().expect("the header is read first");
+    let plain = key
+      .open_in_place(nonce(self.index, last), Aad::from(header), &mut frame)
+      .map_err(|_| corrupt("sealed object failed authentication"))?
+      .len();
+    frame.truncate(plain);
+    self.index += 1;
+    Ok(Bytes::from(frame))
+  }
+
+  pub fn push(&mut self, data: &[u8], out: &mut VecDeque<Bytes>) -> Result<(), StoreError> {
+    self.pending.extend_from_slice(data);
+    if self.key.is_none() && !self.read_header()? {
+      return Ok(());
+    }
+    while self.pending.len() > FRAME + TAG {
+      let rest = self.pending.split_off(FRAME + TAG);
+      let frame = std::mem::replace(&mut self.pending, rest);
+      out.push_back(self.open(frame, false)?);
+    }
+    Ok(())
+  }
+
+  pub fn finish(mut self) -> Result<Bytes, StoreError> {
+    if self.key.is_none() && !self.read_header()? {
+      return Err(corrupt("sealed object is cut off in its header"));
+    }
+    if self.pending.len() < TAG {
+      return Err(corrupt("sealed object is cut off"));
+    }
+    let frame = std::mem::take(&mut self.pending);
+    self.open(frame, true)
+  }
+}
+
+enum Codec {
+  Seal(Option<Sealer>, Option<Bytes>),
+  Open(Option<Opener>),
+}
+
+impl Codec {
+  fn push(&mut self, data: &[u8], out: &mut VecDeque<Bytes>) -> Result<(), StoreError> {
+    match self {
+      Codec::Seal(sealer, header) => {
+        if let Some(header) = header.take() {
+          out.push_back(header);
+        }
+        sealer.as_mut().expect("not finished").push(data, out);
+        Ok(())
+      }
+      Codec::Open(opener) => opener.as_mut().expect("not finished").push(data, out),
+    }
+  }
+
+  fn finish(&mut self, out: &mut VecDeque<Bytes>) -> Result<(), StoreError> {
+    match self {
+      Codec::Seal(sealer, header) => {
+        if let Some(header) = header.take() {
+          out.push_back(header);
+        }
+        out.push_back(sealer.take().expect("finished once").finish());
+        Ok(())
+      }
+      Codec::Open(opener) => {
+        out.push_back(opener.take().expect("finished once").finish()?);
+        Ok(())
+      }
+    }
+  }
+}
+
+fn transform(input: ByteStream, codec: Codec) -> ByteStream {
+  struct State {
+    input: ByteStream,
+    codec: Codec,
+    out: VecDeque<Bytes>,
+    done: bool,
+  }
+  let state = State {
+    input,
+    codec,
+    out: VecDeque::new(),
+    done: false,
+  };
+  Box::pin(futures_util::stream::unfold(
+    state,
+    |mut state| async move {
+      loop {
+        if let Some(chunk) = state.out.pop_front() {
+          if chunk.is_empty() {
+            continue;
+          }
+          return Some((Ok(chunk), state));
+        }
+        if state.done {
+          return None;
+        }
+        let step = match state.input.next().await {
+          Some(Ok(data)) => state.codec.push(&data, &mut state.out),
+          Some(Err(e)) => Err(e),
+          None => {
+            state.done = true;
+            state.codec.finish(&mut state.out)
+          }
+        };
+        if let Err(e) = step {
+          state.done = true;
+          state.out.clear();
+          return Some((Err(e), state));
+        }
+      }
+    },
+  ))
+}
+
+/// `input` sealed under the current key.
+pub fn seal_stream(keyring: &Keyring, input: ByteStream) -> ByteStream {
+  let (sealer, header) = Sealer::new(keyring);
+  transform(input, Codec::Seal(Some(sealer), Some(header)))
+}
+
+/// `input`, a sealed object, opened. A forged, cut or wrongly keyed object
+/// ends the stream with `StoreError::Corrupt`.
+pub fn open_stream(keyring: Arc<Keyring>, input: ByteStream) -> ByteStream {
+  transform(input, Codec::Open(Some(Opener::new(keyring))))
+}
+
+pub fn seal(keyring: &Keyring, plain: &[u8]) -> Vec<u8> {
+  let (mut sealer, header) = Sealer::new(keyring);
+  let mut out = VecDeque::from([header]);
+  sealer.push(plain, &mut out);
+  out.push_back(sealer.finish());
+  out.into_iter().flat_map(|b| b.to_vec()).collect()
+}
+
+pub fn open(keyring: Arc<Keyring>, sealed: &[u8]) -> Result<Vec<u8>, StoreError> {
+  let mut opener = Opener::new(keyring);
+  let mut out = VecDeque::new();
+  opener.push(sealed, &mut out)?;
+  out.push_back(opener.finish()?);
+  Ok(out.into_iter().flat_map(|b| b.to_vec()).collect())
+}
+
+#[cfg(test)]
+#[path = "at_rest_tests.rs"]
+mod tests;
@@ -0,0 +1,347 @@
+use super::*;
+use crate::store::collect;
+
+const K1: &str = "k1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=";
+const K2: &str = "k2:AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE=";
+
+fn keyring(value: &str) -> Arc<Keyring> {
+  Arc::new(Keyring::parse(value).unwrap())
+}
+
+fn sample(len: usize) -> Vec<u8> {
+  (0..len).map(|i| (i * 31 + 7) as u8).collect()
+}
+
+const SIZES: [usize; 9] = [
+  0,
+  1,
+  FRAME - 1,
+  FRAME,
+  FRAME + 1,
+  2 * FRAME - 1,
+  2 * FRAME,
+  2 * FRAME + 1,
+  3 * FRAME + 1234,
+];
+
+fn header_bytes(id: &str) -> usize {
+  header_len(id) as usize
+}
+
+#[test]
+fn every_boundary_size_round_trips_with_the_size_math() {
+  let keys = keyring(K1);
+  for size in SIZES {
+    let plain = sample(size);
+    let sealed = seal(&keys, &plain);
+    assert_eq!(
+      sealed.len() as u64,
+      sealed_len("k1", size as u64),
+      "size {size}"
+    );
+    assert_eq!(
+      plain_len("k1", sealed.len() as u64),
+      Some(size as u64),
+      "size {size}"
+    );
+    assert_eq!(open(keys.clone(), &sealed).unwrap(), plain, "size {size}");
+  }
+}
+
+#[test]
+fn an_empty_object_is_one_empty_final_frame() {
+  let keys = keyring(K1);
+  let sealed = seal(&keys, b"");
+  assert_eq!(sealed.len(), header_bytes("k1") + TAG);
+  assert_eq!(open(keys, &sealed).unwrap(), b"");
+}
+
+#[test]
+fn sizes_no_sealed_object_has_are_refused() {
+  let header = header_len("k1");
+  assert_eq!(plain_len("k1", 0), None);
+  assert_eq!(plain_len("k1", header), None);
+  assert_eq!(plain_len("k1", header + TAG as u64 - 1), None);
+  assert_eq!(plain_len("k1", header + (FRAME + TAG) as u64 + 3), None);
+  assert_eq!(plain_len("k1", header + TAG as u64), Some(0));
+}
+
+#[test]
+fn any_flipped_byte_is_refused() {
+  let keys = keyring(K1);
+  let sealed = seal(&keys, &sample(FRAME + 100));
+  let header = header_bytes("k1");
+  // Magic, version, key id, salt, first frame, first tag, last frame, last tag.
+  for at in [
+    0,
+    4,
+    6,
+    header - 1,
+    header,
+    header + FRAME + TAG - 1,
+    header + FRAME + TAG,
+    sealed.len() - 1,
+  ] {
+    let mut forged = sealed.clone();
+    forged[at] ^= 1;
+    assert!(
+      matches!(open(keys.clone(), &forged), Err(StoreError::Corrupt(_))),
+      "byte {at}"
+    );
+  }
+}
+
+#[test]
+fn truncation_anywhere_is_refused() {
+  let keys = keyring(K1);
+  let sealed = seal(&keys, &sample(2 * FRAME + 10));
+  let header = header_bytes("k1");
+  let frame = FRAME + TAG;
+  for cut in [
+    0,
+    3,
+    header - 1,
+    header,
+    header + 5,
+    header + frame,
+    header + 2 * frame,
+    sealed.len() - 1,
+  ] {
+    assert!(
+      matches!(
+        open(keys.clone(), &sealed[..cut]),
+        Err(StoreError::Corrupt(_))
+      ),
+      "cut at {cut}"
+    );
+  }
+  // Exactly at a frame boundary of an object whose last frame is full.
+  let full = seal(&keys, &sample(2 * FRAME));
+  assert!(open(keys.clone(), &full[..header + frame]).is_err());
+}
+
+#[test]
+fn reordered_dropped_or_appended_frames_are_refused() {
+  let keys = keyring(K1);
+  let sealed = seal(&keys, &sample(3 * FRAME));
+  let header = header_bytes("k1");
+  let frame = FRAME + TAG;
+  let part = |i: usize| &sealed[header + i * frame..header + (i + 1) * frame];
+  let swapped = [&sealed[..header], part(1), part(0), part(2)].concat();
+  assert!(open(keys.clone(), &swapped).is_err());
+  let dropped = [&sealed[..header], part(0), part(2)].concat();
+  assert!(open(keys.clone(), &dropped).is_err());
+  let appended = [&sealed[..], &[0u8; TAG]].concat();
+  assert!(open(keys.clone(), &appended).is_err());
+}
+
+#[test]
+fn a_frame_from_another_object_is_refused() {
+  let keys = keyring(K1);
+  let a = seal(&keys, &sample(10));
+  let b = seal(&keys, &sample(10));
+  let header = header_bytes("k1");
+  let spliced = [&a[..header], &b[header..]].concat();
+  assert!(open(keys, &spliced).is_err());
+}
+
+#[test]
+fn the_wrong_key_or_an_unknown_key_id_is_refused() {
+  let sealed = seal(&keyring(K1), b"secret");
+  let impostor = keyring("k1:AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE=");
+  assert!(matches!(
+    open(impostor, &sealed),
+    Err(StoreError::Corrupt(_))
+  ));
+  let other = keyring(K2);
+  match open(other, &sealed) {
+    Err(StoreError::Corrupt(message)) => assert!(message.contains("'k1'")),
+    other => panic!("{other:?}"),
+  }
+}
+
+#[test]
+fn an_old_key_still_reads_after_rotation_and_new_writes_use_the_first() {
+  let old = seal(&keyring(K1), b"before");
+  let rotated = keyring(&format!("{K2},{K1}"));
+  assert_eq!(open(rotated.clone(), &old).unwrap(), b"before");
+  let new = seal(&rotated, b"after");
+  assert_eq!(&new[6..8], b"k2");
+  assert_eq!(rotated.marker(), "1:k2");
+  assert!(open(keyring(K1), &new).is_err());
+}
+
+#[test]
+fn the_same_plaintext_seals_differently_each_time() {
+  let keys = keyring(K1);
+  assert_ne!(seal(&keys, b"same"), seal(&keys, b"same"));
+}
+
+fn chunked(data: Vec<u8>, sizes: &[usize]) -> crate::store::ByteStream {
+  let mut chunks = Vec::new();
+  let mut at = 0;
+  let mut i = 0;
+  while at < data.len() {
+    let n = sizes[i % sizes.len()].min(data.len() - at);
+    chunks.push(Ok(Bytes::copy_from_slice(&data[at..at + n])));
+    at += n;
+    i += 1;
+  }
+  Box::pin(futures_util::stream::iter(chunks))
+}
+
+#[tokio::test]
+async fn streams_match_the_whole_buffer_functions_for_any_chunking() {
+  let keys = keyring(K1);
+  for size in SIZES {
+    let plain = sample(size);
+    for sizes in [
+      &[1usize][..],
+      &[7, 4093, 65_537],
+      &[FRAME],
+      &[3 * FRAME + 9999],
+    ] {
+      if size > 3 * FRAME && sizes == [1] {
+        continue;
+      }
+      let sealed = collect(seal_stream(&keys, chunked(plain.clone(), sizes)))
+        .await
+        .unwrap();
+      assert_eq!(sealed.len() as u64, sealed_len("k1", size as u64));
+      assert_eq!(open(keys.clone(), &sealed).unwrap(), plain);
+      let opened = collect(open_stream(keys.clone(), chunked(sealed, sizes)))
+        .await
+        .unwrap();
+      assert_eq!(opened, plain, "size {size} chunks {sizes:?}");
+    }
+  }
+}
+
+#[tokio::test]
+async fn a_cut_stream_ends_in_an_error_not_a_short_body() {
+  let keys = keyring(K1);
+  let sealed = seal(&keys, &sample(2 * FRAME + 5));
+  let cut = sealed[..sealed.len() - 3].to_vec();
+  let result = collect(open_stream(keys, chunked(cut, &[1000]))).await;
+  assert!(matches!(result, Err(StoreError::Corrupt(_))));
+}
+
+#[test]
+fn keyrings_parse_strictly() {
+  assert_eq!(keyring(&format!(" {K2} , {K1} ")).current().id(), "k2");
+  let url_safe = keyring("a.b_c-1:AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE");
+  assert_eq!(url_safe.key_count(), 1);
+  for bad in [
+    "",
+    ",",
+    "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=",
+    "k1:AAAA",
+    "k1:not base64!",
+    "k 1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=",
+    ":AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=",
+    &format!("{K1},{K1}"),
+    &format!(
+      "{}:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=",
+      "k".repeat(33)
+    ),
+  ] {
+    assert!(Keyring::parse(bad).is_err(), "{bad}");
+  }
+}
+
+#[test]
+fn markers_name_their_key() {
+  assert_eq!(marker_key_id("1:k1"), Some("k1"));
+  assert_eq!(marker_key_id("2:k1"), None);
+  assert_eq!(marker_key_id("1:"), None);
+}
+
+mod sealed_store {
+  use super::*;
+  use crate::memory_store::MemoryStore;
+  use crate::sealed_store::SealedStore;
+  use crate::store::{Store, Upload, once};
+
+  fn stores() -> (Arc<MemoryStore>, SealedStore) {
+    let raw = Arc::new(MemoryStore::default());
+    let sealed = SealedStore::new(raw.clone(), keyring(K1));
+    (raw, sealed)
+  }
+
+  #[tokio::test]
+  async fn writes_are_sealed_and_read_back_as_written() {
+    let (raw, store) = stores();
+    let metadata = vec![("updated-at".to_string(), "42".to_string())];
+    let plain = sample(FRAME + 3);
+    store
+      .put_object_stream(
+        "a",
+        once(plain.clone()),
+        Upload {
+          length: plain.len() as u64,
+          content_type: Some("application/octet-stream"),
+          metadata: &metadata,
+          if_match: None,
+        },
+      )
+      .await
+      .unwrap();
+    let stored = raw.raw("a").unwrap();
+    assert_ne!(stored.body, plain);
+    assert_eq!(&stored.body[..4], b"DSAR");
+    assert!(
+      stored
+        .metadata
+        .contains(&(MARKER.to_string(), "1:k1".to_string()))
+    );
+
+    let head = store.head_object("a").await.unwrap();
+    assert_eq!(head.content_length, Some(plain.len() as u64));
+    assert_eq!(head.metadata, metadata);
+    let download = store.get_object("a").await.unwrap();
+    assert_eq!(
+      download.head.content_type.as_deref(),
+      Some("application/octet-stream")
+    );
+    assert_eq!(collect(download.body).await.unwrap(), plain);
+
+    store
+      .put_object("t", b"{\"x\":1}".to_vec(), "application/json", &[])
+      .await
+      .unwrap();
+    assert_ne!(raw.get("t").unwrap(), "{\"x\":1}");
+    assert_eq!(store.get_object_text("t").await.unwrap(), "{\"x\":1}");
+  }
+
+  #[tokio::test]
+  async fn plaintext_objects_from_before_read_unchanged() {
+    let (raw, store) = stores();
+    raw.insert(
+      "old",
+      b"DSAR looks sealed but has no marker".to_vec(),
+      vec![("updated-at".to_string(), "7".to_string())],
+      None,
+    );
+    let head = store.head_object("old").await.unwrap();
+    assert_eq!(head.content_length, Some(35));
+    assert_eq!(
+      head.metadata,
+      vec![("updated-at".to_string(), "7".to_string())]
+    );
+    assert_eq!(
+      store.get_object_text("old").await.unwrap(),
+      "DSAR looks sealed but has no marker"
+    );
+  }
+
+  #[tokio::test]
+  async fn copies_stay_readable() {
+    let (_, store) = stores();
+    store
+      .put_object("a", b"x".to_vec(), "text/plain", &[])
+      .await
+      .unwrap();
+    store.copy_object("a", "b").await.unwrap();
+    assert_eq!(store.get_object_text("b").await.unwrap(), "x");
+  }
+}
@@ -17,6 +17,7 @@ pub struct Config {
  pub s3_force_path_style: Option<String>,
  pub s3_public_endpoint: Option<String>,
  pub port: Option<String>,
+  pub at_rest_key: Option<String>,
 }

 fn env(name: &str) -> Option<String> {
@@ -41,6 +42,7 @@ impl Config {
      s3_force_path_style: env("S3_FORCE_PATH_STYLE"),
      s3_public_endpoint: env("S3_PUBLIC_ENDPOINT"),
      port: env("PORT"),
+      at_rest_key: env("SYNC_AT_REST_KEY"),
    }
  }

@@ -6,7 +6,7 @@ use std::sync::Arc;
 use std::time::Duration;

 use bytes::Bytes;
-use http_body_util::combinators::BoxBody;
+use http_body_util::combinators::UnsyncBoxBody;
 use http_body_util::{BodyExt, Full, StreamBody};
 use hyper::body::{Frame, Incoming};
 use hyper::http::request::Parts;
@@ -19,7 +19,8 @@ use crate::jsval::Body;
 use crate::log;
 use crate::sync::SyncService;

-pub type ResBody = BoxBody<Bytes, Infallible>;
+pub type BoxError = Box<dyn std::error::Error + Send + Sync>;
+pub type ResBody = UnsyncBoxBody<Bytes, BoxError>;

 /// The JSON body limit (`50mb` in body-parser terms).
 pub const BODY_LIMIT: usize = 50 * 1024 * 1024;
@@ -32,10 +33,10 @@ pub struct AppState {
  pub account_url: Option<String>,
 }

-fn full(bytes: impl Into<Bytes>) -> ResBody {
+pub fn full(bytes: impl Into<Bytes>) -> ResBody {
  Full::new(bytes.into())
    .map_err(|never| match never {})
-    .boxed()
+    .boxed_unsync()
 }

 fn json_response(status: u16, body: &Value) -> Response<ResBody> {
@@ -71,7 +72,7 @@ fn header_value(parts: &Parts, name: impl header::AsHeaderName) -> Option<String
 /// Reads and discards what is left of a refused body, as Node did, so the
 /// client finishes sending and reads the answer instead of a reset
 /// connection. Bounded, so a client cannot keep the server reading.
-async fn drain(mut body: Incoming) {
+pub async fn drain(mut body: Incoming) {
  const MAX: usize = 128 * 1024 * 1024;
  let mut read = 0usize;
  let _ = tokio::time::timeout(Duration::from_secs(30), async {
@@ -199,6 +200,13 @@ pub async fn handle(
  request: Request<Incoming>,
 ) -> Result<Response<ResBody>, Infallible> {
  let (parts, body) = request.into_parts();
+  // A transfer's body is the object, streamed: it must not be read as JSON.
+  if let Some(signer) = state.sync.transfer()
+    && route_path(parts.uri.path()) == crate::transfer::PATH
+  {
+    let response = crate::transfer::serve(state.sync.store(), signer, &parts, body).await;
+    return Ok(with_cors(response));
+  }
  let body = match read_json_body(&parts, body).await {
    Ok(body) => body,
    Err(response) => return Ok(*response),
@@ -253,7 +261,7 @@ async fn route(state: &Arc<AppState>, parts: &Parts, body: Body) -> Response<Res
        Ok(ctx) => ctx,
        Err(response) => return *response,
      };
-      match object_route_call(&state.sync, name, &body, &ctx).await {
+      match object_route_call(&state.sync, name, &body, &ctx, &origin(state, parts)).await {
        Ok(value) => json_response(200, &value),
        Err(error) => error_response(error),
      }
@@ -277,20 +285,42 @@ const OBJECT_ROUTES: [&str; 12] = [
  "snapshots/restore-files",
 ];

+/// How devices reach this server, for the transfer URLs of a sealed store:
+/// `SYNC_PUBLIC_URL`, or the address this request came to.
+fn origin(state: &AppState, parts: &Parts) -> String {
+  if let Some(url) = &state.account_url {
+    return url.clone();
+  }
+  let first = |name: &str| {
+    header_value(parts, name)
+      .and_then(|v| v.split(',').next().map(|s| s.trim().to_string()))
+      .filter(|v| !v.is_empty())
+  };
+  let scheme = match first("x-forwarded-proto").as_deref() {
+    Some("https") => "https",
+    _ => "http",
+  };
+  let host = first("x-forwarded-host")
+    .or_else(|| first("host"))
+    .unwrap_or_else(|| "localhost".to_string());
+  format!("{scheme}://{host}")
+}
+
 async fn object_route_call(
  sync: &SyncService,
  name: &str,
  body: &Body,
  ctx: &UserContext,
+  origin: &str,
 ) -> Result<Value, ApiError> {
  match name {
    "stat" => sync.stat(body, ctx).await,
-    "presign-upload" => sync.presign_upload(body, ctx).await,
-    "presign-download" => sync.presign_download(body, ctx).await,
+    "presign-upload" => sync.presign_upload(body, ctx, origin).await,
+    "presign-download" => sync.presign_download(body, ctx, origin).await,
    "delete" => sync.delete(body, ctx).await,
    "list" => sync.list(body, ctx).await,
-    "presign-upload-batch" => sync.presign_upload_batch(body, ctx).await,
-    "presign-download-batch" => sync.presign_download_batch(body, ctx).await,
+    "presign-upload-batch" => sync.presign_upload_batch(body, ctx, origin).await,
+    "presign-download-batch" => sync.presign_download_batch(body, ctx, origin).await,
    "delete-prefix" => sync.delete_prefix(body, ctx).await,
    "snapshots/list" => sync.list_snapshots(body, ctx).await,
    "snapshots/create" => sync.create_snapshot(body, ctx).await,
@@ -356,7 +386,7 @@ fn subscribe(state: &AppState, ctx: UserContext) -> Response<ResBody> {
  let stream = futures_util::stream::unfold(rx, |mut rx| async move {
    rx.recv()
      .await
-      .map(|bytes| (Ok::<_, Infallible>(Frame::data(bytes)), rx))
+      .map(|bytes| (Ok::<_, BoxError>(Frame::data(bytes)), rx))
  });
  Response::builder()
    .header(header::CONTENT_TYPE, "text/event-stream")
@@ -368,7 +398,7 @@ fn subscribe(state: &AppState, ctx: UserContext) -> Response<ResBody> {
    .header(header::PRAGMA, "no-cache")
    .header(header::EXPIRES, "0")
    .header("x-accel-buffering", "no")
-    .body(StreamBody::new(stream).boxed())
+    .body(StreamBody::new(stream).boxed_unsync())
    .expect("static response parts are valid")
 }

@@ -3,6 +3,7 @@
 //! presigned URLs, so profile files go straight between devices and storage.

 pub mod accounts;
+pub mod at_rest;
 pub mod auth;
 pub mod backend;
 pub mod config;
@@ -11,13 +12,18 @@ pub mod http;
 pub mod jsval;
 pub mod jwt;
 pub mod log;
+#[cfg(test)]
+mod memory_store;
+pub mod migrate;
 pub mod net;
 pub mod s3;
+pub mod sealed_store;
 pub mod sigv4;
 pub mod store;
 pub mod subscribe;
 pub mod sync;
 pub mod time;
+pub mod transfer;

 use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
 use std::sync::Arc;
@@ -28,13 +34,17 @@ use hyper_util::rt::{TokioIo, TokioTimer};
 use tokio::net::TcpListener;

 use crate::accounts::{DonutAccounts, normalize_sync_server_url};
+use crate::at_rest::Keyring;
 use crate::auth::Guard;
 use crate::backend::Backend;
 use crate::config::Config;
 use crate::http::AppState;
 use crate::s3::{Location, Presigner, S3Store};
+use crate::sealed_store::SealedStore;
 use crate::sigv4::Credentials;
+use crate::store::Store;
 use crate::sync::{SyncOptions, SyncService};
+use crate::transfer::TransferSigner;

 const DEFAULT_PORT: u16 = 3929;
 /// Shipped placeholders that must never guard a real server.
@@ -74,6 +84,9 @@ pub fn validate_env(config: &Config) -> Result<(), String> {
          .to_string(),
      );
  }
+  if let Some(value) = &config.at_rest_key {
+    Keyring::parse(value)?;
+  }
  Ok(())
 }

@@ -83,49 +96,95 @@ fn require(value: &Option<String>, name: &str) -> Result<String, String> {
    .ok_or_else(|| format!("Required environment variable {name} is not set"))
 }

+struct Storage {
+  endpoint: String,
+  bucket: String,
+  region: String,
+  credentials: Credentials,
+  force_path_style: bool,
+}
+
+impl Storage {
+  // No defaults for storage: a misconfigured server must not start pointed at
+  // an unintended or public-default backend.
+  fn from_config(config: &Config) -> Result<Self, String> {
+    Ok(Self {
+      endpoint: require(&config.s3_endpoint, "S3_ENDPOINT")?,
+      bucket: require(&config.s3_bucket, "S3_BUCKET")?,
+      region: config
+        .s3_region
+        .clone()
+        .unwrap_or_else(|| "us-east-1".to_string()),
+      credentials: Credentials {
+        access_key_id: require(&config.s3_access_key_id, "S3_ACCESS_KEY_ID")?,
+        secret_access_key: require(&config.s3_secret_access_key, "S3_SECRET_ACCESS_KEY")?,
+      },
+      force_path_style: config.s3_force_path_style.as_deref() != Some("false"),
+    })
+  }
+
+  fn store(&self) -> Result<S3Store, String> {
+    Ok(S3Store::new(
+      Location::parse(&self.endpoint, &self.bucket, self.force_path_style)?,
+      self.credentials.clone(),
+      self.region.clone(),
+    ))
+  }
+}
+
+/// The bucket itself, without sealing: for the migration.
+pub fn raw_store(config: &Config) -> Result<S3Store, String> {
+  Storage::from_config(config)?.store()
+}
+
 /// Everything the server needs, connected and checked: storage reachable and
 /// the bucket created if it was missing.
 pub async fn build_state(config: &Config) -> Result<Arc<AppState>, String> {
-  // No defaults for storage: a misconfigured server must not start pointed at
-  // an unintended or public-default backend.
-  let endpoint = require(&config.s3_endpoint, "S3_ENDPOINT")?;
-  let region = config
-    .s3_region
-    .clone()
-    .unwrap_or_else(|| "us-east-1".to_string());
-  let access_key_id = require(&config.s3_access_key_id, "S3_ACCESS_KEY_ID")?;
-  let secret_access_key = require(&config.s3_secret_access_key, "S3_SECRET_ACCESS_KEY")?;
-  let force_path_style = config.s3_force_path_style.as_deref() != Some("false");
-  let bucket = require(&config.s3_bucket, "S3_BUCKET")?;
-  let credentials = Credentials {
-    access_key_id,
-    secret_access_key,
-  };
-
+  let storage = Storage::from_config(config)?;
  // Presigned URLs go to a desktop on another machine, so they must name a
  // host that machine can reach. Signing is bound to the host, so they are
  // signed against the public address rather than rewritten after signing.
  let public_endpoint = config
    .s3_public_endpoint
    .clone()
-    .unwrap_or_else(|| endpoint.clone());
-  let store = S3Store::new(
-    Location::parse(&endpoint, &bucket, force_path_style)?,
-    credentials.clone(),
-    region.clone(),
-  );
+    .unwrap_or_else(|| storage.endpoint.clone());
+  let s3: Arc<dyn Store> = Arc::new(storage.store()?);
+  let keyring = config
+    .at_rest_key
+    .as_deref()
+    .map(Keyring::parse)
+    .transpose()?
+    .map(Arc::new);
+  let (store, transfer): (Arc<dyn Store>, _) = match &keyring {
+    Some(keyring) => {
+      crate::log::log(
+        "SyncService",
+        &format!(
+          "Encryption at rest on key={} keys={}",
+          keyring.current().id(),
+          keyring.key_count()
+        ),
+      );
+      (
+        Arc::new(SealedStore::new(s3, keyring.clone())),
+        Some(TransferSigner::new(keyring)),
+      )
+    }
+    None => (s3, None),
+  };
  let presigner = Presigner::new(
-    Location::parse(&public_endpoint, &bucket, force_path_style)?,
-    credentials,
-    region,
+    Location::parse(&public_endpoint, &storage.bucket, storage.force_path_style)?,
+    storage.credentials,
+    storage.region,
  );

  let backend = config
    .backend()
    .map(|(url, key)| Arc::new(Backend::new(url, key)));
  let sync = Arc::new(SyncService::new(
-    Arc::new(store),
+    store,
    SyncOptions {
+      transfer,
      presigner,
      public_endpoint,
      public_endpoint_configured: config.s3_public_endpoint.is_some(),
@@ -240,6 +299,10 @@ pub async fn run() -> i32 {
  // A `.env` in the working directory fills in what the environment lacks.
  let _ = dotenvy::dotenv();
  let config = Config::from_env();
+  let args: Vec<String> = std::env::args().skip(1).collect();
+  if args.first().map(String::as_str) == Some(migrate::COMMAND) {
+    return migrate::run(&config, &args[1..]).await;
+  }
  if let Err(message) = validate_env(&config) {
    eprintln!("{message}");
    return 1;
@@ -0,0 +1,237 @@
+//! A bucket in memory, for unit tests. Hooks stand in for HEAD and LIST where
+//! a test needs answers the bucket itself would not give.
+
+use std::collections::BTreeMap;
+use std::sync::Mutex;
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+use async_trait::async_trait;
+use md5::{Digest, Md5};
+
+use crate::store::{
+  ByteStream, Download, HeadOutput, ListInput, ListObject, ListOutput, Store, StoreError, Upload,
+  collect, once,
+};
+
+pub type HeadHook = Box<dyn Fn(&str) -> Result<HeadOutput, StoreError> + Send + Sync>;
+pub type ListHook = Box<dyn Fn(&ListInput) -> ListOutput + Send + Sync>;
+
+#[derive(Clone)]
+pub struct Object {
+  pub body: Vec<u8>,
+  pub metadata: Vec<(String, String)>,
+  pub content_type: Option<String>,
+}
+
+#[derive(Default)]
+pub struct MemoryStore {
+  pub objects: Mutex<BTreeMap<String, Object>>,
+  pub calls: AtomicUsize,
+  pub head_hook: Option<HeadHook>,
+  pub list_hook: Option<ListHook>,
+}
+
+fn etag(body: &[u8]) -> String {
+  format!("\"{}\"", hex::encode(Md5::digest(body)))
+}
+
+impl MemoryStore {
+  fn call(&self) {
+    self.calls.fetch_add(1, Ordering::SeqCst);
+  }
+
+  pub fn set(&self, key: &str, body: &str) {
+    self.insert(key, body.as_bytes().to_vec(), Vec::new(), None);
+  }
+
+  pub fn insert(
+    &self,
+    key: &str,
+    body: Vec<u8>,
+    metadata: Vec<(String, String)>,
+    content_type: Option<String>,
+  ) {
+    self.objects.lock().unwrap().insert(
+      key.to_string(),
+      Object {
+        body,
+        metadata,
+        content_type,
+      },
+    );
+  }
+
+  pub fn get(&self, key: &str) -> Option<String> {
+    self
+      .raw(key)
+      .map(|o| String::from_utf8_lossy(&o.body).into_owned())
+  }
+
+  pub fn raw(&self, key: &str) -> Option<Object> {
+    self.objects.lock().unwrap().get(key).cloned()
+  }
+
+  pub fn keys(&self) -> Vec<String> {
+    self.objects.lock().unwrap().keys().cloned().collect()
+  }
+
+  fn head(&self, key: &str) -> Result<HeadOutput, StoreError> {
+    let object = self.raw(key).ok_or(StoreError::NotFound)?;
+    let mut metadata = object.metadata.clone();
+    metadata.sort();
+    Ok(HeadOutput {
+      last_modified_ms: Some(0),
+      content_length: Some(object.body.len() as u64),
+      metadata,
+      etag: Some(etag(&object.body)),
+      content_type: object.content_type,
+    })
+  }
+}
+
+#[async_trait]
+impl Store for MemoryStore {
+  async fn head_bucket(&self) -> Result<(), StoreError> {
+    self.call();
+    Ok(())
+  }
+
+  async fn create_bucket(&self) -> Result<(), StoreError> {
+    self.call();
+    Ok(())
+  }
+
+  async fn head_object(&self, key: &str) -> Result<HeadOutput, StoreError> {
+    self.call();
+    if let Some(hook) = &self.head_hook {
+      return hook(key);
+    }
+    self.head(key)
+  }
+
+  async fn get_object_text(&self, key: &str) -> Result<String, StoreError> {
+    self.call();
+    self.get(key).ok_or(StoreError::NotFound)
+  }
+
+  async fn put_object(
+    &self,
+    key: &str,
+    body: Vec<u8>,
+    content_type: &str,
+    metadata: &[(String, String)],
+  ) -> Result<(), StoreError> {
+    self.call();
+    self.insert(key, body, metadata.to_vec(), Some(content_type.to_string()));
+    Ok(())
+  }
+
+  async fn delete_object(&self, key: &str) -> Result<(), StoreError> {
+    self.call();
+    self.objects.lock().unwrap().remove(key);
+    Ok(())
+  }
+
+  async fn delete_objects(&self, keys: &[String]) -> Result<(), StoreError> {
+    self.call();
+    let mut objects = self.objects.lock().unwrap();
+    for key in keys {
+      objects.remove(key);
+    }
+    Ok(())
+  }
+
+  async fn copy_object(&self, from: &str, to: &str) -> Result<(), StoreError> {
+    self.call();
+    let mut objects = self.objects.lock().unwrap();
+    let object = objects.get(from).cloned().ok_or(StoreError::NotFound)?;
+    objects.insert(to.to_string(), object);
+    Ok(())
+  }
+
+  async fn list_objects_v2(&self, input: &ListInput) -> Result<ListOutput, StoreError> {
+    self.call();
+    if let Some(hook) = &self.list_hook {
+      return Ok(hook(input));
+    }
+    let prefix = input.prefix.clone().unwrap_or_default();
+    let max_keys: usize = input
+      .max_keys
+      .as_deref()
+      .and_then(|m| m.parse().ok())
+      .unwrap_or(1000);
+    let after = input
+      .continuation_token
+      .clone()
+      .or_else(|| input.start_after.clone());
+    let objects = self.objects.lock().unwrap();
+    let mut out = ListOutput::default();
+    for (key, object) in objects.range(prefix.clone()..) {
+      if !key.starts_with(&prefix) {
+        break;
+      }
+      if after.as_ref().is_some_and(|after| key <= after) {
+        continue;
+      }
+      if let Some(delimiter) = &input.delimiter
+        && let Some(at) = key[prefix.len()..].find(delimiter.as_str())
+      {
+        let common = key[..prefix.len() + at + delimiter.len()].to_string();
+        if !out.common_prefixes.contains(&common) {
+          out.common_prefixes.push(common);
+        }
+        continue;
+      }
+      if out.contents.len() == max_keys {
+        out.is_truncated = true;
+        out.next_continuation_token = out.contents.last().map(|o| o.key.clone());
+        break;
+      }
+      out.contents.push(ListObject {
+        key: key.clone(),
+        last_modified_ms: Some(0),
+        size: Some(object.body.len() as u64),
+      });
+    }
+    Ok(out)
+  }
+
+  async fn get_object(&self, key: &str) -> Result<Download, StoreError> {
+    self.call();
+    let head = self.head(key)?;
+    let body = self.raw(key).ok_or(StoreError::NotFound)?.body;
+    Ok(Download {
+      head,
+      body: once(body),
+    })
+  }
+
+  async fn put_object_stream(
+    &self,
+    key: &str,
+    body: ByteStream,
+    upload: Upload<'_>,
+  ) -> Result<(), StoreError> {
+    self.call();
+    let body = collect(body).await?;
+    if body.len() as u64 != upload.length {
+      return Err(StoreError::Transport("length mismatch".to_string()));
+    }
+    if let Some(expected) = upload.if_match
+      && self.raw(key).map(|o| etag(&o.body)).as_deref() != Some(expected)
+    {
+      return Err(StoreError::Service {
+        status: 412,
+        code: "PreconditionFailed".to_string(),
+        message: String::new(),
+      });
+    }
+    self.insert(
+      key,
+      body,
+      upload.metadata.to_vec(),
+      upload.content_type.map(str::to_string),
+    );
+    Ok(())
+  }
+}
@@ -0,0 +1,638 @@
+//! `donut-sync migrate-at-rest`: seals the plaintext objects already in the
+//! bucket, and re-seals objects under an older key, while the server runs.
+//!
+//! Run it after the server runs with `SYNC_AT_REST_KEY` and the S3 URLs it
+//! handed out before have expired (one hour), so no device writes plaintext
+//! behind it. A rerun skips what is sealed under the current key.
+//!
+//! A device can write an object while it is re-sealed. Where the store honors
+//! `If-Match` on PUT, the write is conditional on the ETag read. Elsewhere the
+//! object is sealed into a temporary key outside every namespace, its ETag is
+//! checked again, and a server-side copy puts it in place, so the window is
+//! one HEAD and one copy, whatever the object's size.
+
+use std::path::PathBuf;
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
+
+use futures_util::StreamExt;
+
+use crate::at_rest::{self, Keyring, MARKER};
+use crate::config::Config;
+use crate::log;
+use crate::sealed_store::sealed_by;
+use crate::store::{ByteStream, ListInput, Store, StoreError, Upload};
+
+pub const COMMAND: &str = "migrate-at-rest";
+const CONTEXT: &str = "Migrate";
+/// Temporary sealed copies. No client prefix matches it.
+pub const TMP_PREFIX: &str = ".donut-sync-at-rest-tmp/";
+const PROGRESS_EVERY: Duration = Duration::from_secs(10);
+/// Tries per object: storage has transient failures.
+const ATTEMPTS: u32 = 3;
+
+const USAGE: &str = "Usage: donut-sync migrate-at-rest [--dry-run] [--prefix P] [--concurrency N] [--max-mib-per-sec N] [--checkpoint FILE]
+
+Seals every plaintext object in S3_BUCKET with the first key of SYNC_AT_REST_KEY,
+and re-seals objects under its other keys. Safe to stop and run again.
+
+  --dry-run            Count what would change; write nothing.
+  --prefix P           Only keys under P.
+  --concurrency N      Objects at a time (default 4).
+  --max-mib-per-sec N  Read at most N MiB/s (default unlimited).
+  --checkpoint FILE    Resume after the key in FILE, and keep it current.";
+
+#[derive(Debug, Default, Clone)]
+pub struct Options {
+  pub dry_run: bool,
+  pub prefix: Option<String>,
+  pub concurrency: usize,
+  pub bytes_per_sec: Option<u64>,
+  pub checkpoint: Option<PathBuf>,
+}
+
+fn parse(args: &[String]) -> Result<Options, String> {
+  let mut options = Options {
+    concurrency: 4,
+    ..Options::default()
+  };
+  let mut args = args.iter();
+  while let Some(arg) = args.next() {
+    let mut value = |name: &str| {
+      args
+        .next()
+        .cloned()
+        .ok_or_else(|| format!("{name} needs a value"))
+    };
+    match arg.as_str() {
+      "--dry-run" => options.dry_run = true,
+      "--prefix" => options.prefix = Some(value(arg)?),
+      "--concurrency" => {
+        options.concurrency = value(arg)?
+          .parse()
+          .ok()
+          .filter(|n| (1..=64).contains(n))
+          .ok_or("--concurrency must be 1-64")?;
+      }
+      "--max-mib-per-sec" => {
+        let mib: u64 = value(arg)?
+          .parse()
+          .ok()
+          .filter(|n| *n > 0)
+          .ok_or("--max-mib-per-sec must be a positive whole number")?;
+        options.bytes_per_sec = Some(mib * 1024 * 1024);
+      }
+      "--checkpoint" => options.checkpoint = Some(PathBuf::from(value(arg)?)),
+      _ => return Err(format!("Unknown argument {arg}")),
+    }
+  }
+  Ok(options)
+}
+
+pub async fn run(config: &Config, args: &[String]) -> i32 {
+  if args.iter().any(|a| a == "--help" || a == "-h") {
+    println!("{USAGE}");
+    return 0;
+  }
+  let options = match parse(args) {
+    Ok(options) => options,
+    Err(message) => {
+      eprintln!("{message}\n\n{USAGE}");
+      return 2;
+    }
+  };
+  let Some(keys) = config.at_rest_key.as_deref() else {
+    eprintln!("SYNC_AT_REST_KEY is not set");
+    return 1;
+  };
+  let keyring = match Keyring::parse(keys) {
+    Ok(keyring) => Arc::new(keyring),
+    Err(message) => {
+      eprintln!("{message}");
+      return 1;
+    }
+  };
+  let store: Arc<dyn Store> = match crate::raw_store(config) {
+    Ok(store) => Arc::new(store),
+    Err(message) => {
+      eprintln!("{message}");
+      return 1;
+    }
+  };
+  match migrate(store, keyring, &options).await {
+    Ok(report) if report.failed == 0 => 0,
+    Ok(_) => 1,
+    Err(message) => {
+      log::error(CONTEXT, &format!("Migration stopped err=\"{message}\""));
+      1
+    }
+  }
+}
+
+#[derive(Debug, Default, Clone, PartialEq)]
+pub struct Report {
+  pub scanned: u64,
+  /// Already under the current key.
+  pub current: u64,
+  /// Plaintext objects sealed (or, in a dry run, to seal).
+  pub sealed: u64,
+  pub sealed_bytes: u64,
+  /// Objects under an older key re-sealed (or to re-seal).
+  pub rotated: u64,
+  pub rotated_bytes: u64,
+  /// Written or deleted by someone else meanwhile; left as they are.
+  pub changed: u64,
+  pub failed: u64,
+}
+
+enum Outcome {
+  Current,
+  Sealed(u64),
+  Rotated(u64),
+  Changed,
+  Failed(String),
+}
+
+/// Reads at most `per_sec` bytes a second, across every worker.
+struct RateLimit {
+  per_sec: u64,
+  state: Mutex<(Instant, u64)>,
+}
+
+impl RateLimit {
+  async fn take(&self, bytes: u64) {
+    let wait = {
+      let mut state = self.state.lock().unwrap();
+      state.1 += bytes;
+      let due = Duration::from_secs_f64(state.1 as f64 / self.per_sec as f64);
+      due.saturating_sub(state.0.elapsed())
+    };
+    if !wait.is_zero() {
+      tokio::time::sleep(wait).await;
+    }
+  }
+}
+
+fn throttled(body: ByteStream, limit: Option<Arc<RateLimit>>) -> ByteStream {
+  match limit {
+    None => body,
+    Some(limit) => Box::pin(body.then(move |chunk| {
+      let limit = limit.clone();
+      async move {
+        if let Ok(bytes) = &chunk {
+          limit.take(bytes.len() as u64).await;
+        }
+        chunk
+      }
+    })),
+  }
+}
+
+struct Migration {
+  store: Arc<dyn Store>,
+  keyring: Arc<Keyring>,
+  dry_run: bool,
+  conditional: bool,
+  limit: Option<Arc<RateLimit>>,
+}
+
+fn precondition_failed(error: &StoreError) -> bool {
+  matches!(error, StoreError::Service { status: 412, .. })
+}
+
+/// Whether the store refuses a PUT whose `If-Match` names another ETag.
+pub async fn honors_if_match(store: &dyn Store) -> Result<bool, StoreError> {
+  let probe = format!("{TMP_PREFIX}probe-{}", uuid::Uuid::new_v4().simple());
+  store
+    .put_object(&probe, b"a".to_vec(), "text/plain", &[])
+    .await?;
+  let outcome = store
+    .put_object_stream(
+      &probe,
+      crate::store::once(b"b".to_vec()),
+      Upload {
+        length: 1,
+        content_type: Some("text/plain"),
+        metadata: &[],
+        if_match: Some("\"00000000000000000000000000000000\""),
+      },
+    )
+    .await;
+  let _ = store.delete_object(&probe).await;
+  match outcome {
+    Ok(()) => Ok(false),
+    Err(e) if precondition_failed(&e) => Ok(true),
+    // Some stores (Backblaze B2) refuse the header instead of honoring it.
+    Err(StoreError::Service { .. }) => Ok(false),
+    Err(e) => Err(e),
+  }
+}
+
+impl Migration {
+  async fn object(&self, key: &str) -> Outcome {
+    let mut attempt = 0;
+    loop {
+      attempt += 1;
+      match self.try_object(key).await {
+        Ok(outcome) => return outcome,
+        Err(StoreError::NotFound) => return Outcome::Changed,
+        Err(e) if precondition_failed(&e) => return Outcome::Changed,
+        Err(e @ StoreError::Corrupt(_)) => return Outcome::Failed(e.to_string()),
+        Err(e) if attempt >= ATTEMPTS => return Outcome::Failed(e.to_string()),
+        Err(_) => tokio::time::sleep(Duration::from_millis(500 << attempt)).await,
+      }
+    }
+  }
+
+  async fn try_object(&self, key: &str) -> Result<Outcome, StoreError> {
+    let head = self.store.head_object(key).await?;
+    let current = self.keyring.current().id();
+    let old_key = match sealed_by(&head) {
+      Some(id) if id == current => return Ok(Outcome::Current),
+      Some(id) => Some(id.to_string()),
+      None => None,
+    };
+    let stored = head.content_length.unwrap_or(0);
+    let plain = match &old_key {
+      Some(id) => at_rest::plain_len(id, stored).ok_or_else(|| {
+        StoreError::Corrupt(format!("sealed object has an invalid size {stored}"))
+      })?,
+      None => stored,
+    };
+    let done = |plain| match old_key {
+      Some(_) => Outcome::Rotated(plain),
+      None => Outcome::Sealed(plain),
+    };
+    if self.dry_run {
+      return Ok(done(plain));
+    }
+    let etag = head
+      .etag
+      .clone()
+      .ok_or_else(|| StoreError::Transport("the store gave no ETag".to_string()))?;
+    let download = self.store.get_object(key).await?;
+    if download.head.etag.as_deref() != Some(etag.as_str()) {
+      return Ok(Outcome::Changed);
+    }
+    let mut body = throttled(download.body, self.limit.clone());
+    if old_key.is_some() {
+      body = at_rest::open_stream(self.keyring.clone(), body);
+    }
+    let sealed = at_rest::seal_stream(&self.keyring, body);
+    let mut metadata: Vec<(String, String)> = head
+      .metadata
+      .iter()
+      .filter(|(name, _)| name != MARKER)
+      .cloned()
+      .collect();
+    metadata.push((MARKER.to_string(), self.keyring.marker()));
+    let upload = Upload {
+      length: at_rest::sealed_len(current, plain),
+      content_type: head.content_type.as_deref(),
+      metadata: &metadata,
+      if_match: None,
+    };
+    if self.conditional {
+      self
+        .store
+        .put_object_stream(
+          key,
+          sealed,
+          Upload {
+            if_match: Some(&etag),
+            ..upload
+          },
+        )
+        .await?;
+      return Ok(done(plain));
+    }
+    let tmp = format!("{TMP_PREFIX}{}", uuid::Uuid::new_v4().simple());
+    let placed = async {
+      self.store.put_object_stream(&tmp, sealed, upload).await?;
+      if self.store.head_object(key).await?.etag.as_deref() != Some(etag.as_str()) {
+        return Ok(false);
+      }
+      self.store.copy_object(&tmp, key).await?;
+      Ok::<_, StoreError>(true)
+    }
+    .await;
+    let _ = self.store.delete_object(&tmp).await;
+    Ok(if placed? {
+      done(plain)
+    } else {
+      Outcome::Changed
+    })
+  }
+}
+
+fn read_checkpoint(path: &PathBuf) -> Result<Option<String>, String> {
+  match std::fs::read_to_string(path) {
+    Ok(text) => Ok(Some(text.trim_end_matches('\n').to_string()).filter(|k| !k.is_empty())),
+    Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
+    Err(e) => Err(format!("Cannot read checkpoint {}: {e}", path.display())),
+  }
+}
+
+fn write_checkpoint(path: &PathBuf, key: &str) -> Result<(), String> {
+  let tmp = path.with_extension("tmp");
+  std::fs::write(&tmp, format!("{key}\n"))
+    .and_then(|()| std::fs::rename(&tmp, path))
+    .map_err(|e| format!("Cannot write checkpoint {}: {e}", path.display()))
+}
+
+fn summary(report: &Report, elapsed: Duration) -> String {
+  format!(
+    "scanned={} current={} sealed={} sealed_bytes={} rotated={} rotated_bytes={} changed={} failed={} elapsed_ms={}",
+    report.scanned,
+    report.current,
+    report.sealed,
+    report.sealed_bytes,
+    report.rotated,
+    report.rotated_bytes,
+    report.changed,
+    report.failed,
+    elapsed.as_millis()
+  )
+}
+
+/// Walks the bucket in key order. `Err` stops the walk: a listing or the
+/// checkpoint failed. Objects that fail are counted, and keep the checkpoint
+/// from moving past them.
+pub async fn migrate(
+  store: Arc<dyn Store>,
+  keyring: Arc<Keyring>,
+  options: &Options,
+) -> Result<Report, String> {
+  let started = Instant::now();
+  let conditional = if options.dry_run {
+    false
+  } else {
+    honors_if_match(store.as_ref())
+      .await
+      .map_err(|e| format!("Cannot write to the bucket: {e}"))?
+  };
+  let start_after = match &options.checkpoint {
+    Some(path) => read_checkpoint(path)?,
+    None => None,
+  };
+  log::log(
+    CONTEXT,
+    &format!(
+      "Migration started key={} dry_run={} conditional_put={conditional} concurrency={} max_bytes_per_sec={} resume_after={}",
+      keyring.current().id(),
+      options.dry_run,
+      options.concurrency,
+      options.bytes_per_sec.unwrap_or(0),
+      start_after.as_deref().unwrap_or("-"),
+    ),
+  );
+  let migration = Migration {
+    store: store.clone(),
+    keyring,
+    dry_run: options.dry_run,
+    conditional,
+    limit: options.bytes_per_sec.map(|per_sec| {
+      Arc::new(RateLimit {
+        per_sec,
+        state: Mutex::new((Instant::now(), 0)),
+      })
+    }),
+  };
+  let mut report = Report::default();
+  let mut token: Option<String> = None;
+  let mut last_progress = Instant::now();
+  let mut checkpoint_held = false;
+  loop {
+    let page = store
+      .list_objects_v2(&ListInput {
+        prefix: options.prefix.clone(),
+        max_keys: Some("1000".to_string()),
+        continuation_token: token.clone(),
+        delimiter: None,
+        start_after: if token.is_none() {
+          start_after.clone()
+        } else {
+          None
+        },
+      })
+      .await
+      .map_err(|e| format!("Listing failed: {e}"))?;
+    let keys: Vec<String> = page
+      .contents
+      .into_iter()
+      .map(|object| object.key)
+      .filter(|key| !key.starts_with(TMP_PREFIX))
+      .collect();
+    let outcomes: Vec<(String, Outcome)> = futures_util::stream::iter(keys.iter())
+      .map(|key| async { (key.clone(), migration.object(key).await) })
+      .buffer_unordered(options.concurrency)
+      .collect()
+      .await;
+    for (key, outcome) in outcomes {
+      report.scanned += 1;
+      match outcome {
+        Outcome::Current => report.current += 1,
+        Outcome::Sealed(bytes) => {
+          report.sealed += 1;
+          report.sealed_bytes += bytes;
+        }
+        Outcome::Rotated(bytes) => {
+          report.rotated += 1;
+          report.rotated_bytes += bytes;
+        }
+        Outcome::Changed => report.changed += 1,
+        Outcome::Failed(err) => {
+          report.failed += 1;
+          checkpoint_held = true;
+          log::warn(
+            CONTEXT,
+            &format!("Object not migrated key={key} err=\"{err}\""),
+          );
+        }
+      }
+    }
+    if let (Some(path), Some(last), false, false) = (
+      &options.checkpoint,
+      keys.last(),
+      checkpoint_held,
+      options.dry_run,
+    ) {
+      write_checkpoint(path, last)?;
+    }
+    token = page.next_continuation_token.filter(|_| page.is_truncated);
+    if token.is_none() {
+      break;
+    }
+    if last_progress.elapsed() >= PROGRESS_EVERY {
+      last_progress = Instant::now();
+      log::log(
+        CONTEXT,
+        &format!(
+          "Migration progress {} last_key={}",
+          summary(&report, started.elapsed()),
+          keys.last().map_or("-", String::as_str)
+        ),
+      );
+    }
+  }
+  let line = format!(
+    "{} {}",
+    if options.dry_run {
+      "Migration dry run done"
+    } else {
+      "Migration done"
+    },
+    summary(&report, started.elapsed())
+  );
+  if report.failed > 0 {
+    log::warn(CONTEXT, &line);
+  } else {
+    log::log(CONTEXT, &line);
+  }
+  Ok(report)
+}
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+  use crate::memory_store::MemoryStore;
+  use crate::sealed_store::SealedStore;
+
+  const K1: &str = "k1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=";
+  const K2: &str = "k2:AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE=";
+
+  fn options() -> Options {
+    Options {
+      concurrency: 3,
+      ..Options::default()
+    }
+  }
+
+  fn meta(value: &str) -> Vec<(String, String)> {
+    vec![("updated-at".to_string(), value.to_string())]
+  }
+
+  #[tokio::test]
+  async fn seals_plaintext_keeps_metadata_and_is_idempotent() {
+    let raw = Arc::new(MemoryStore::default());
+    let big: Vec<u8> = (0..at_rest::FRAME * 2 + 9).map(|i| i as u8).collect();
+    raw.insert(
+      "users/u/profiles/p/files/a",
+      big.clone(),
+      meta("5"),
+      Some("image/png".into()),
+    );
+    raw.insert("users/u/proxies/x.json", b"{}".to_vec(), Vec::new(), None);
+    raw.insert("users/u/empty", Vec::new(), Vec::new(), None);
+    let keyring = Arc::new(Keyring::parse(K1).unwrap());
+
+    let dry = migrate(
+      raw.clone(),
+      keyring.clone(),
+      &Options {
+        dry_run: true,
+        ..options()
+      },
+    )
+    .await
+    .unwrap();
+    assert_eq!(
+      (dry.scanned, dry.sealed, dry.sealed_bytes),
+      (3, 3, big.len() as u64 + 2)
+    );
+    assert_eq!(raw.raw("users/u/proxies/x.json").unwrap().body, b"{}");
+
+    let report = migrate(raw.clone(), keyring.clone(), &options())
+      .await
+      .unwrap();
+    assert_eq!((report.sealed, report.failed, report.changed), (3, 0, 0));
+    let sealed = SealedStore::new(raw.clone(), keyring.clone());
+    let head = sealed
+      .head_object("users/u/profiles/p/files/a")
+      .await
+      .unwrap();
+    assert_eq!(head.metadata, meta("5"));
+    assert_eq!(head.content_type.as_deref(), Some("image/png"));
+    assert_eq!(head.content_length, Some(big.len() as u64));
+    let body = sealed
+      .get_object("users/u/profiles/p/files/a")
+      .await
+      .unwrap()
+      .body;
+    assert_eq!(crate::store::collect(body).await.unwrap(), big);
+    assert_eq!(sealed.get_object_text("users/u/empty").await.unwrap(), "");
+    assert!(raw.keys().iter().all(|k| !k.starts_with(TMP_PREFIX)));
+
+    let again = migrate(raw.clone(), keyring, &options()).await.unwrap();
+    assert_eq!((again.current, again.sealed), (3, 0));
+  }
+
+  #[tokio::test]
+  async fn rotates_objects_under_an_older_key() {
+    let raw = Arc::new(MemoryStore::default());
+    let old = Arc::new(Keyring::parse(K1).unwrap());
+    SealedStore::new(raw.clone(), old)
+      .put_object("a", b"hello".to_vec(), "text/plain", &meta("1"))
+      .await
+      .unwrap();
+    let rotated = Arc::new(Keyring::parse(&format!("{K2},{K1}")).unwrap());
+    let report = migrate(raw.clone(), rotated.clone(), &options())
+      .await
+      .unwrap();
+    assert_eq!((report.rotated, report.rotated_bytes), (1, 5));
+    let only_new = Arc::new(Keyring::parse(K2).unwrap());
+    let store = SealedStore::new(raw.clone(), only_new);
+    assert_eq!(store.get_object_text("a").await.unwrap(), "hello");
+    assert_eq!(store.head_object("a").await.unwrap().metadata, meta("1"));
+  }
+
+  #[tokio::test]
+  async fn resumes_after_the_checkpoint() {
+    let raw = Arc::new(MemoryStore::default());
+    for i in 0..2500 {
+      raw.insert(&format!("k{i:05}"), b"x".to_vec(), Vec::new(), None);
+    }
+    let keyring = Arc::new(Keyring::parse(K1).unwrap());
+    let dir = std::env::temp_dir().join(format!("donut-migrate-{}", uuid::Uuid::new_v4()));
+    std::fs::create_dir_all(&dir).unwrap();
+    let path = dir.join("checkpoint");
+    std::fs::write(&path, "k01999\n").unwrap();
+    let report = migrate(
+      raw.clone(),
+      keyring,
+      &Options {
+        checkpoint: Some(path.clone()),
+        ..options()
+      },
+    )
+    .await
+    .unwrap();
+    assert_eq!(report.sealed, 500);
+    assert_eq!(std::fs::read_to_string(&path).unwrap(), "k02499\n");
+    assert_eq!(raw.raw("k00000").unwrap().body, b"x");
+    std::fs::remove_dir_all(dir).unwrap();
+  }
+
+  #[tokio::test]
+  async fn the_probe_sees_whether_if_match_is_honored() {
+    // MemoryStore refuses a PUT whose If-Match names another ETag.
+    let raw = Arc::new(MemoryStore::default());
+    assert!(honors_if_match(raw.as_ref()).await.unwrap());
+    assert!(raw.keys().is_empty(), "the probe cleans up");
+  }
+
+  #[test]
+  fn arguments_parse() {
+    let args: Vec<String> = ["--dry-run", "--prefix", "users/", "--max-mib-per-sec", "8"]
+      .iter()
+      .map(|s| s.to_string())
+      .collect();
+    let options = parse(&args).unwrap();
+    assert!(options.dry_run);
+    assert_eq!(options.prefix.as_deref(), Some("users/"));
+    assert_eq!(options.bytes_per_sec, Some(8 * 1024 * 1024));
+    assert_eq!(options.concurrency, 4);
+    assert!(parse(&["--concurrency".to_string(), "0".to_string()]).is_err());
+    assert!(parse(&["--bogus".to_string()]).is_err());
+  }
+}
@@ -10,7 +10,9 @@ use reqwest::Method;

 use crate::jsval::js_number;
 use crate::sigv4::{self, Credentials};
-use crate::store::{HeadOutput, ListInput, ListObject, ListOutput, Store, StoreError};
+use crate::store::{
+  ByteStream, Download, HeadOutput, ListInput, ListObject, ListOutput, Store, StoreError, Upload,
+};
 use crate::time;

 const MAX_ATTEMPTS: u32 = 3;
@@ -377,16 +379,16 @@ impl S3Store {
    }
  }

-  async fn attempt(
+  /// A signed request, without its body.
+  fn request(
    &self,
    method: &Method,
    path: &str,
    query: &[(String, String)],
    headers: &[(String, String)],
-    body: &[u8],
-  ) -> Result<Reply, StoreError> {
+    payload_hash: String,
+  ) -> reqwest::RequestBuilder {
    let amz_date = time::compact_stamp(time::now_ms());
-    let payload_hash = sigv4::sha256_hex(body);
    let mut signed = vec![
      ("host".to_string(), self.location.request_host()),
      ("x-amz-content-sha256".to_string(), payload_hash.clone()),
@@ -414,6 +416,18 @@ impl S3Store {
    for (name, value) in signed.iter().filter(|(name, _)| name != "host") {
      request = request.header(name.as_str(), value.as_str());
    }
+    request
+  }
+
+  async fn attempt(
+    &self,
+    method: &Method,
+    path: &str,
+    query: &[(String, String)],
+    headers: &[(String, String)],
+    body: &[u8],
+  ) -> Result<Reply, StoreError> {
+    let mut request = self.request(method, path, query, headers, sigv4::sha256_hex(body));
    if !body.is_empty() || *method == Method::PUT || *method == Method::POST {
      request = request.body(body.to_vec());
    }
@@ -487,6 +501,44 @@ impl S3Store {
  }
 }

+/// What a HEAD or GET says about the object.
+fn head_of(headers: &reqwest::header::HeaderMap) -> HeadOutput {
+  let header = |name: &str| headers.get(name).map(|v| latin1(v.as_bytes()));
+  let mut metadata: Vec<(String, String)> = headers
+    .iter()
+    .filter_map(|(name, value)| {
+      name
+        .as_str()
+        .strip_prefix("x-amz-meta-")
+        .map(|n| (n.to_string(), latin1(value.as_bytes())))
+    })
+    .collect();
+  metadata.sort_by(|a, b| a.0.cmp(&b.0));
+  HeadOutput {
+    last_modified_ms: header("last-modified").and_then(|v| time::parse_http_date_ms(&v)),
+    content_length: header("content-length").and_then(|v| v.trim().parse().ok()),
+    metadata,
+    etag: header("etag"),
+    content_type: header("content-type"),
+  }
+}
+
+/// A refusal, read from a streamed response.
+async fn refusal(response: reqwest::Response) -> StoreError {
+  let status = response.status().as_u16();
+  let headers = response.headers().clone();
+  let body = response
+    .bytes()
+    .await
+    .map(|b| b.to_vec())
+    .unwrap_or_default();
+  error_of(&Reply {
+    status,
+    headers,
+    body,
+  })
+}
+
 fn transport_message(error: &reqwest::Error) -> String {
  let mut message = error.to_string();
  let mut source = std::error::Error::source(error);
@@ -521,24 +573,7 @@ impl Store for S3Store {
    let reply = self
      .expect_success(Method::HEAD, &path, &[], &[], &[])
      .await?;
-    let header = |name: &str| reply.headers.get(name).map(|v| latin1(v.as_bytes()));
-    let mut metadata: Vec<(String, String)> = reply
-      .headers
-      .iter()
-      .filter_map(|(name, value)| {
-        name
-          .as_str()
-          .strip_prefix("x-amz-meta-")
-          .map(|n| (n.to_string(), latin1(value.as_bytes())))
-      })
-      .collect();
-    metadata.sort_by(|a, b| a.0.cmp(&b.0));
-    Ok(HeadOutput {
-      last_modified_ms: header("last-modified").and_then(|v| time::parse_http_date_ms(&v)),
-      content_length: header("content-length").and_then(|v| v.trim().parse().ok()),
-      metadata,
-      etag: header("etag"),
-    })
+    Ok(head_of(&reply.headers))
  }

  async fn get_object_text(&self, key: &str) -> Result<String, StoreError> {
@@ -554,9 +589,13 @@ impl Store for S3Store {
    key: &str,
    body: Vec<u8>,
    content_type: &str,
+    metadata: &[(String, String)],
  ) -> Result<(), StoreError> {
    let path = self.location.object_path(key);
-    let headers = [("content-type".to_string(), content_type.to_string())];
+    let mut headers = vec![("content-type".to_string(), content_type.to_string())];
+    for (name, value) in metadata {
+      headers.push((format!("x-amz-meta-{name}"), value.clone()));
+    }
    self
      .expect_success(Method::PUT, &path, &[], &headers, &body)
      .await
@@ -640,12 +679,82 @@ impl Store for S3Store {
    if let Some(prefix) = &input.prefix {
      query.push(("prefix".to_string(), prefix.clone()));
    }
+    if let Some(start_after) = &input.start_after {
+      query.push(("start-after".to_string(), start_after.clone()));
+    }
    let path = self.location.bucket_path();
    let reply = self
      .expect_success(Method::GET, &path, &query, &[], &[])
      .await?;
    parse_list(&reply.body)
  }
+
+  async fn get_object(&self, key: &str) -> Result<Download, StoreError> {
+    let path = self.location.object_path(key);
+    let mut attempt = 0;
+    let response = loop {
+      attempt += 1;
+      let sent = self
+        .request(&Method::GET, &path, &[], &[], sigv4::sha256_hex(&[]))
+        .send()
+        .await;
+      let retry = match &sent {
+        Err(_) => true,
+        Ok(response) => matches!(response.status().as_u16(), 429 | 500 | 502 | 503 | 504),
+      };
+      if !retry || attempt >= MAX_ATTEMPTS {
+        break sent.map_err(|e| StoreError::Transport(transport_message(&e)))?;
+      }
+      tokio::time::sleep(RETRY_BASE.saturating_mul(1 << attempt)).await;
+    };
+    if !response.status().is_success() {
+      return Err(refusal(response).await);
+    }
+    let head = head_of(response.headers());
+    let body = futures_util::StreamExt::map(response.bytes_stream(), |chunk| {
+      chunk.map_err(|e| StoreError::Transport(transport_message(&e)))
+    });
+    Ok(Download {
+      head,
+      body: Box::pin(body),
+    })
+  }
+
+  /// One attempt: a stream cannot be sent twice.
+  async fn put_object_stream(
+    &self,
+    key: &str,
+    body: ByteStream,
+    upload: Upload<'_>,
+  ) -> Result<(), StoreError> {
+    let path = self.location.object_path(key);
+    let mut headers = vec![("content-length".to_string(), upload.length.to_string())];
+    if let Some(content_type) = upload.content_type {
+      headers.push(("content-type".to_string(), content_type.to_string()));
+    }
+    if let Some(etag) = upload.if_match {
+      headers.push(("if-match".to_string(), etag.to_string()));
+    }
+    for (name, value) in upload.metadata {
+      headers.push((format!("x-amz-meta-{name}"), value.clone()));
+    }
+    let response = self
+      .request(
+        &Method::PUT,
+        &path,
+        &[],
+        &headers,
+        sigv4::UNSIGNED_PAYLOAD.to_string(),
+      )
+      .body(reqwest::Body::wrap_stream(body))
+      .send()
+      .await
+      .map_err(|e| StoreError::Transport(transport_message(&e)))?;
+    if !response.status().is_success() {
+      return Err(refusal(response).await);
+    }
+    Ok(())
+  }
 }

 #[cfg(test)]
@@ -0,0 +1,133 @@
+//! The store as the routes see it when `SYNC_AT_REST_KEY` is set: writes are
+//! sealed, reads opened, and plaintext objects from before still read as they
+//! are. HEAD answers with the plaintext size and without the marker.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+
+use crate::at_rest::{self, Keyring, MARKER};
+use crate::store::{
+  ByteStream, Download, HeadOutput, ListInput, ListOutput, Store, StoreError, Upload,
+};
+
+pub struct SealedStore {
+  inner: Arc<dyn Store>,
+  keyring: Arc<Keyring>,
+}
+
+/// The key id a stored object is sealed under; `None` for plaintext.
+pub fn sealed_by(head: &HeadOutput) -> Option<&str> {
+  head.metadata_value(MARKER).and_then(at_rest::marker_key_id)
+}
+
+/// `head` as the plaintext object would have it.
+fn reveal(mut head: HeadOutput) -> Result<HeadOutput, StoreError> {
+  if let Some(id) = sealed_by(&head) {
+    let id = id.to_string();
+    head.content_length = match head.content_length {
+      Some(sealed) => Some(at_rest::plain_len(&id, sealed).ok_or_else(|| {
+        StoreError::Corrupt(format!("sealed object has an invalid size {sealed}"))
+      })?),
+      None => None,
+    };
+    head.metadata.retain(|(name, _)| name != MARKER);
+  }
+  Ok(head)
+}
+
+impl SealedStore {
+  pub fn new(inner: Arc<dyn Store>, keyring: Arc<Keyring>) -> Self {
+    Self { inner, keyring }
+  }
+
+  fn with_marker(&self, metadata: &[(String, String)]) -> Vec<(String, String)> {
+    let mut out = metadata.to_vec();
+    out.push((MARKER.to_string(), self.keyring.marker()));
+    out
+  }
+}
+
+#[async_trait]
+impl Store for SealedStore {
+  async fn head_bucket(&self) -> Result<(), StoreError> {
+    self.inner.head_bucket().await
+  }
+
+  async fn create_bucket(&self) -> Result<(), StoreError> {
+    self.inner.create_bucket().await
+  }
+
+  async fn head_object(&self, key: &str) -> Result<HeadOutput, StoreError> {
+    reveal(self.inner.head_object(key).await?)
+  }
+
+  async fn get_object_text(&self, key: &str) -> Result<String, StoreError> {
+    let body = crate::store::collect(self.get_object(key).await?.body).await?;
+    Ok(String::from_utf8_lossy(&body).into_owned())
+  }
+
+  async fn put_object(
+    &self,
+    key: &str,
+    body: Vec<u8>,
+    content_type: &str,
+    metadata: &[(String, String)],
+  ) -> Result<(), StoreError> {
+    let sealed = at_rest::seal(&self.keyring, &body);
+    self
+      .inner
+      .put_object(key, sealed, content_type, &self.with_marker(metadata))
+      .await
+  }
+
+  async fn delete_object(&self, key: &str) -> Result<(), StoreError> {
+    self.inner.delete_object(key).await
+  }
+
+  async fn delete_objects(&self, keys: &[String]) -> Result<(), StoreError> {
+    self.inner.delete_objects(keys).await
+  }
+
+  /// Sealed objects copy as they are: the frames do not name their key.
+  async fn copy_object(&self, from: &str, to: &str) -> Result<(), StoreError> {
+    self.inner.copy_object(from, to).await
+  }
+
+  async fn list_objects_v2(&self, input: &ListInput) -> Result<ListOutput, StoreError> {
+    self.inner.list_objects_v2(input).await
+  }
+
+  async fn get_object(&self, key: &str) -> Result<Download, StoreError> {
+    let download = self.inner.get_object(key).await?;
+    let sealed = sealed_by(&download.head).is_some();
+    let head = reveal(download.head)?;
+    let body = if sealed {
+      at_rest::open_stream(self.keyring.clone(), download.body)
+    } else {
+      download.body
+    };
+    Ok(Download { head, body })
+  }
+
+  async fn put_object_stream(
+    &self,
+    key: &str,
+    body: ByteStream,
+    upload: Upload<'_>,
+  ) -> Result<(), StoreError> {
+    let metadata = self.with_marker(upload.metadata);
+    self
+      .inner
+      .put_object_stream(
+        key,
+        at_rest::seal_stream(&self.keyring, body),
+        Upload {
+          length: at_rest::sealed_len(self.keyring.current().id(), upload.length),
+          metadata: &metadata,
+          ..upload
+        },
+      )
+      .await
+  }
+}
@@ -1,6 +1,12 @@
 //! The object storage operations the server uses.

+use std::pin::Pin;
+
 use async_trait::async_trait;
+use bytes::Bytes;
+use futures_util::Stream;
+
+pub type ByteStream = Pin<Box<dyn Stream<Item = Result<Bytes, StoreError>> + Send>>;

 #[derive(Debug, Clone, Default)]
 pub struct HeadOutput {
@@ -9,6 +15,7 @@ pub struct HeadOutput {
  /// User metadata: lowercase names without `x-amz-meta-`, in response order.
  pub metadata: Vec<(String, String)>,
  pub etag: Option<String>,
+  pub content_type: Option<String>,
 }

 impl HeadOutput {
@@ -27,6 +34,7 @@ pub struct ListInput {
  pub max_keys: Option<String>,
  pub continuation_token: Option<String>,
  pub delimiter: Option<String>,
+  pub start_after: Option<String>,
 }

 #[derive(Debug, Clone)]
@@ -56,6 +64,8 @@ pub enum StoreError {
  },
  /// The store could not be reached, or answered with something unreadable.
  Transport(String),
+  /// A sealed object that does not open: forged, cut off, or under an unknown key.
+  Corrupt(String),
 }

 impl StoreError {
@@ -83,6 +93,7 @@ impl std::fmt::Display for StoreError {
        }
      }
      StoreError::Transport(message) => write!(f, "{message}"),
+      StoreError::Corrupt(message) => write!(f, "{message}"),
    }
  }
 }
@@ -100,6 +111,7 @@ pub trait Store: Send + Sync {
    key: &str,
    body: Vec<u8>,
    content_type: &str,
+    metadata: &[(String, String)],
  ) -> Result<(), StoreError>;
  async fn delete_object(&self, key: &str) -> Result<(), StoreError>;
  /// Deletes up to 1000 keys in one request; per-key failures are ignored.
@@ -107,4 +119,44 @@ pub trait Store: Send + Sync {
  /// Server-side copy, metadata included.
  async fn copy_object(&self, from: &str, to: &str) -> Result<(), StoreError>;
  async fn list_objects_v2(&self, input: &ListInput) -> Result<ListOutput, StoreError>;
+  /// The object's body as it arrives, with what a HEAD would say.
+  async fn get_object(&self, key: &str) -> Result<Download, StoreError>;
+  /// Writes `upload.length` bytes from `body`. With `if_match`, the store
+  /// refuses (412) when the object's ETag differs, where it supports that.
+  async fn put_object_stream(
+    &self,
+    key: &str,
+    body: ByteStream,
+    upload: Upload<'_>,
+  ) -> Result<(), StoreError>;
+}
+
+pub struct Download {
+  pub head: HeadOutput,
+  pub body: ByteStream,
+}
+
+#[derive(Clone, Copy)]
+pub struct Upload<'a> {
+  pub length: u64,
+  pub content_type: Option<&'a str>,
+  /// Names without `x-amz-meta-`.
+  pub metadata: &'a [(String, String)],
+  pub if_match: Option<&'a str>,
+}
+
+/// All of a stream, in memory.
+pub async fn collect(mut body: ByteStream) -> Result<Vec<u8>, StoreError> {
+  use futures_util::StreamExt;
+  let mut out = Vec::new();
+  while let Some(chunk) = body.next().await {
+    out.extend_from_slice(&chunk?);
+  }
+  Ok(out)
+}
+
+pub fn once(bytes: Vec<u8>) -> ByteStream {
+  Box::pin(futures_util::stream::once(
+    async move { Ok(Bytes::from(bytes)) },
+  ))
 }
@@ -116,6 +116,7 @@ async fn poll_once(
          max_keys: Some("1000".to_string()),
          continuation_token: token.clone(),
          delimiter: None,
+          start_after: None,
        })
        .await
      {
@@ -17,6 +17,7 @@ use crate::log;
 use crate::s3::Presigner;
 use crate::store::{HeadOutput, ListInput, Store, StoreError};
 use crate::time;
+use crate::transfer::TransferSigner;

 const CONTEXT: &str = "SyncService";

@@ -78,6 +79,12 @@ fn clamp_expires_in(requested: Field) -> f64 {
  value.min(MAX_PRESIGN_EXPIRES_IN)
 }

+/// When a URL issued at `now` for `expires_in` seconds stops working, as
+/// `expiresAt` reports it.
+fn expires_ms(now: i64, expires_in: f64) -> i64 {
+  (now as f64 + expires_in * 1000.0).trunc() as i64
+}
+
 /// `new Date(ms).toISOString()` for a fractional millisecond time.
 fn iso_at(ms: f64) -> String {
  time::iso_from_ms(ms.trunc() as i64)
@@ -147,6 +154,7 @@ fn group_denied() -> ApiError {
 pub struct SyncService {
  store: Arc<dyn Store>,
  presigner: Presigner,
+  transfer: Option<TransferSigner>,
  public_endpoint: String,
  public_endpoint_configured: bool,
  cloud: bool,
@@ -156,6 +164,8 @@ pub struct SyncService {

 pub struct SyncOptions {
  pub presigner: Presigner,
+  /// Set when the store is sealed at rest: transfers go through this server.
+  pub transfer: Option<TransferSigner>,
  pub public_endpoint: String,
  pub public_endpoint_configured: bool,
  /// `SYNC_JWT_PUBLIC_KEY` is set: Donut's own deployment.
@@ -168,6 +178,7 @@ impl SyncService {
    Self {
      store,
      presigner: options.presigner,
+      transfer: options.transfer,
      public_endpoint: options.public_endpoint,
      public_endpoint_configured: options.public_endpoint_configured,
      cloud: options.cloud,
@@ -180,6 +191,31 @@ impl SyncService {
    &self.store
  }

+  pub fn transfer(&self) -> Option<&TransferSigner> {
+    self.transfer.as_ref()
+  }
+
+  fn put_url(
+    &self,
+    origin: &str,
+    key: &str,
+    expires_in: f64,
+    metadata: &[(String, String)],
+    now: i64,
+  ) -> String {
+    match &self.transfer {
+      Some(signer) => signer.url(origin, "PUT", key, expires_ms(now, expires_in), metadata),
+      None => self.presigner.put_url(key, expires_in, metadata, now),
+    }
+  }
+
+  fn get_url(&self, origin: &str, key: &str, expires_in: f64, now: i64) -> String {
+    match &self.transfer {
+      Some(signer) => signer.url(origin, "GET", key, expires_ms(now, expires_in), &[]),
+      None => self.presigner.get_url(key, expires_in, now),
+    }
+  }
+
  // --- startup and probes ----------------------------------------------------

  pub async fn ensure_bucket_exists(&self) -> Result<(), String> {
@@ -202,14 +238,18 @@ impl SyncService {
  /// The address clients are sent to for transfers, for `/readyz`. Withheld
  /// in cloud mode: `/readyz` is public, and a managed deployment should not
  /// publish its storage host.
+  /// Withheld too when transfers come here: devices never reach storage.
  pub fn diagnostic_storage_endpoint(&self) -> Option<&str> {
-    (!self.cloud).then_some(self.public_endpoint.as_str())
+    (!self.cloud && self.transfer.is_none()).then_some(self.public_endpoint.as_str())
  }

  /// A single-label host (`minio`, `s3`) resolves only inside a container
  /// network, so every presigned URL built from it fails for the desktop
  /// client while the server's own S3 calls succeed. Say it once at boot.
  pub fn warn_if_presign_endpoint_is_server_only(&self) {
+    if self.transfer.is_some() {
+      return;
+    }
    let Ok(url) = url::Url::parse(&self.public_endpoint) else {
      return;
    };
@@ -427,7 +467,7 @@ impl SyncService {
      })
      .to_string();
      if let Err(e) = store
-        .put_object(&key, body.into_bytes(), "application/json")
+        .put_object(&key, body.into_bytes(), "application/json", &[])
        .await
      {
        log::warn(CONTEXT, &format!("Manifest bump failed for {key}: {e}"));
@@ -448,7 +488,12 @@ impl SyncService {
    }
  }

-  pub async fn presign_upload(&self, body: &Body, ctx: &UserContext) -> Result<Value, ApiError> {
+  pub async fn presign_upload(
+    &self,
+    body: &Body,
+    ctx: &UserContext,
+    origin: &str,
+  ) -> Result<Value, ApiError> {
    let raw_key = body.field("key")?;
    let key = self.scope_key(ctx, raw_key);
    self.validate_write_access(ctx, key.as_deref())?;
@@ -463,7 +508,8 @@ impl SyncService {
    let expires_in = clamp_expires_in(body.field("expiresIn")?);
    let now = time::now_ms();
    let expires_at = iso_at(now as f64 + expires_in * 1000.0);
-    let url = self.presigner.put_url(
+    let url = self.put_url(
+      origin,
      require_key(&key)?,
      expires_in,
      metadata.as_deref().unwrap_or(&[]),
@@ -484,14 +530,19 @@ impl SyncService {
    Ok(Value::Object(out))
  }

-  pub async fn presign_download(&self, body: &Body, ctx: &UserContext) -> Result<Value, ApiError> {
+  pub async fn presign_download(
+    &self,
+    body: &Body,
+    ctx: &UserContext,
+    origin: &str,
+  ) -> Result<Value, ApiError> {
    let key = self.scope_key(ctx, body.field("key")?);
    self.validate_key_access(ctx, key.as_deref())?;
    self.assert_reachable(ctx, key.as_deref()).await?;
    let expires_in = clamp_expires_in(body.field("expiresIn")?);
    let now = time::now_ms();
    let expires_at = iso_at(now as f64 + expires_in * 1000.0);
-    let url = self.presigner.get_url(require_key(&key)?, expires_in, now);
+    let url = self.get_url(origin, require_key(&key)?, expires_in, now);
    Ok(json!({ "url": url, "expiresAt": expires_at }))
  }

@@ -538,6 +589,7 @@ impl SyncService {
          require_key(&tombstone_key)?,
          Value::Object(record).to_string().into_bytes(),
          "application/json",
+          &[],
        )
        .await
        .map_err(store_error)?;
@@ -571,6 +623,7 @@ impl SyncService {
        }),
        continuation_token: token.to_key(),
        delimiter: None,
+        start_after: None,
      })
      .await
      .map_err(store_error)?;
@@ -615,6 +668,7 @@ impl SyncService {
    &self,
    body: &Body,
    ctx: &UserContext,
+    origin: &str,
  ) -> Result<Value, ApiError> {
    let items = body
      .field("items")?
@@ -655,9 +709,7 @@ impl SyncService {
    .await?;
    let mut out_items = Vec::with_capacity(keys.len());
    for (raw, key) in raw_keys.iter().zip(&keys) {
-      let url = self
-        .presigner
-        .put_url(require_key(key)?, expires_in, &[], now);
+      let url = self.put_url(origin, require_key(key)?, expires_in, &[], now);
      let mut item = Map::new();
      if let Field::Value(value) = raw {
        item.insert("key".to_string(), (*value).clone());
@@ -695,6 +747,7 @@ impl SyncService {
    &self,
    body: &Body,
    ctx: &UserContext,
+    origin: &str,
  ) -> Result<Value, ApiError> {
    let raw_keys = body
      .field("keys")?
@@ -724,7 +777,7 @@ impl SyncService {
    .await?;
    let mut items = Vec::with_capacity(keys.len());
    for (raw, key) in raw_keys.iter().zip(&keys) {
-      let url = self.presigner.get_url(require_key(key)?, expires_in, now);
+      let url = self.get_url(origin, require_key(key)?, expires_in, now);
      items.push(json!({ "key": raw, "url": url, "expiresAt": expires_at }));
    }
    Ok(json!({ "items": items }))
@@ -762,6 +815,7 @@ impl SyncService {
          max_keys: Some("1000".to_string()),
          continuation_token: token.clone(),
          delimiter: None,
+          start_after: None,
        })
        .await
        .map_err(store_error)?;
@@ -794,6 +848,7 @@ impl SyncService {
          require_key(&tombstone_key)?,
          Value::Object(record).to_string().into_bytes(),
          "application/json",
+          &[],
        )
        .await
        .map_err(store_error)?;
@@ -837,6 +892,7 @@ impl SyncService {
          max_keys: None,
          continuation_token: token.clone(),
          delimiter: None,
+          start_after: None,
        })
        .await
        .map_err(store_error)?;
@@ -977,6 +1033,7 @@ impl SyncService {
        &format!("{target}snapshot.json"),
        snapshot.to_string().into_bytes(),
        "application/json",
+        &[],
      )
      .await
      .map_err(store_error)?;
@@ -1108,6 +1165,7 @@ impl SyncService {
          max_keys: Some("1000".to_string()),
          continuation_token: token.clone(),
          delimiter: Some("/".to_string()),
+          start_after: None,
        })
        .await
        .map_err(store_error)?;
@@ -1125,6 +1183,7 @@ impl SyncService {
            max_keys: Some("1".to_string()),
            continuation_token: None,
            delimiter: None,
+            start_after: None,
          })
          .await
          .map_err(store_error)?;
@@ -1160,6 +1219,7 @@ impl SyncService {
            max_keys: Some("1000".to_string()),
            continuation_token: token.clone(),
            delimiter: None,
+            start_after: None,
          })
          .await
          .map_err(store_error)?;
@@ -1187,6 +1247,7 @@ impl SyncService {
          &format!("{user_prefix}tombstones/profiles/{id}"),
          record.to_string().into_bytes(),
          "application/json",
+          &[],
        )
        .await
        .map_err(store_error)?;
@@ -1290,6 +1351,7 @@ async fn stored_profile_ids(store: &dyn Store, prefix: &str) -> Result<HashSet<S
        max_keys: Some("1000".to_string()),
        continuation_token: token.clone(),
        delimiter: Some("/".to_string()),
+        start_after: None,
      })
      .await
      .map_err(store_error)?;
@@ -1,165 +1,15 @@
 use std::collections::BTreeMap;
-use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::atomic::Ordering;

-use async_trait::async_trait;
 use serde_json::json;

 use super::*;
+use crate::memory_store::MemoryStore;
 use crate::store::{ListObject, ListOutput};

+const ORIGIN: &str = "http://sync.test";
 const OWNER_PREFIX: &str = "users/99999999-8888-7777-6666-555555555555/";

-type HeadHook = Box<dyn Fn(&str) -> Result<HeadOutput, StoreError> + Send + Sync>;
-type ListHook = Box<dyn Fn(&ListInput) -> ListOutput + Send + Sync>;
-
-#[derive(Clone)]
-struct Object {
-  body: Vec<u8>,
-  metadata: Vec<(String, String)>,
-}
-
-/// A bucket in memory. Hooks stand in for HEAD and LIST where a test needs
-/// answers the bucket itself would not give.
-#[derive(Default)]
-struct MemoryStore {
-  objects: Mutex<BTreeMap<String, Object>>,
-  calls: AtomicUsize,
-  head_hook: Option<HeadHook>,
-  list_hook: Option<ListHook>,
-}
-
-impl MemoryStore {
-  fn call(&self) {
-    self.calls.fetch_add(1, Ordering::SeqCst);
-  }
-
-  fn set(&self, key: &str, body: &str) {
-    self.objects.lock().unwrap().insert(
-      key.to_string(),
-      Object {
-        body: body.as_bytes().to_vec(),
-        metadata: Vec::new(),
-      },
-    );
-  }
-
-  fn get(&self, key: &str) -> Option<String> {
-    self
-      .objects
-      .lock()
-      .unwrap()
-      .get(key)
-      .map(|o| String::from_utf8_lossy(&o.body).into_owned())
-  }
-
-  fn keys(&self) -> Vec<String> {
-    self.objects.lock().unwrap().keys().cloned().collect()
-  }
-}
-
-#[async_trait]
-impl Store for MemoryStore {
-  async fn head_bucket(&self) -> Result<(), StoreError> {
-    self.call();
-    Ok(())
-  }
-
-  async fn create_bucket(&self) -> Result<(), StoreError> {
-    self.call();
-    Ok(())
-  }
-
-  async fn head_object(&self, key: &str) -> Result<HeadOutput, StoreError> {
-    self.call();
-    if let Some(hook) = &self.head_hook {
-      return hook(key);
-    }
-    let objects = self.objects.lock().unwrap();
-    let object = objects.get(key).ok_or(StoreError::NotFound)?;
-    Ok(HeadOutput {
-      last_modified_ms: Some(0),
-      content_length: Some(object.body.len() as u64),
-      metadata: object.metadata.clone(),
-      etag: Some(format!("\"{}\"", object.body.len())),
-    })
-  }
-
-  async fn get_object_text(&self, key: &str) -> Result<String, StoreError> {
-    self.call();
-    self.get(key).ok_or(StoreError::NotFound)
-  }
-
-  async fn put_object(
-    &self,
-    key: &str,
-    body: Vec<u8>,
-    _content_type: &str,
-  ) -> Result<(), StoreError> {
-    self.call();
-    self.objects.lock().unwrap().insert(
-      key.to_string(),
-      Object {
-        body,
-        metadata: Vec::new(),
-      },
-    );
-    Ok(())
-  }
-
-  async fn delete_object(&self, key: &str) -> Result<(), StoreError> {
-    self.call();
-    self.objects.lock().unwrap().remove(key);
-    Ok(())
-  }
-
-  async fn delete_objects(&self, keys: &[String]) -> Result<(), StoreError> {
-    self.call();
-    let mut objects = self.objects.lock().unwrap();
-    for key in keys {
-      objects.remove(key);
-    }
-    Ok(())
-  }
-
-  async fn copy_object(&self, from: &str, to: &str) -> Result<(), StoreError> {
-    self.call();
-    let mut objects = self.objects.lock().unwrap();
-    let object = objects.get(from).cloned().ok_or(StoreError::NotFound)?;
-    objects.insert(to.to_string(), object);
-    Ok(())
-  }
-
-  async fn list_objects_v2(&self, input: &ListInput) -> Result<ListOutput, StoreError> {
-    self.call();
-    if let Some(hook) = &self.list_hook {
-      return Ok(hook(input));
-    }
-    let prefix = input.prefix.clone().unwrap_or_default();
-    let objects = self.objects.lock().unwrap();
-    let mut out = ListOutput::default();
-    for (key, object) in objects.range(prefix.clone()..) {
-      if !key.starts_with(&prefix) {
-        break;
-      }
-      if let Some(delimiter) = &input.delimiter
-        && let Some(at) = key[prefix.len()..].find(delimiter.as_str())
-      {
-        let common = key[..prefix.len() + at + delimiter.len()].to_string();
-        if !out.common_prefixes.contains(&common) {
-          out.common_prefixes.push(common);
-        }
-        continue;
-      }
-      out.contents.push(ListObject {
-        key: key.clone(),
-        last_modified_ms: Some(0),
-        size: Some(object.body.len() as u64),
-      });
-    }
-    Ok(out)
-  }
-}
-
 fn service_with(store: Arc<MemoryStore>) -> SyncService {
  let location = crate::s3::Location::parse("http://s3.invalid:9000", "test-bucket", true).unwrap();
  let presigner = Presigner::new(
@@ -174,6 +24,7 @@ fn service_with(store: Arc<MemoryStore>) -> SyncService {
    store,
    SyncOptions {
      presigner,
+      transfer: None,
      public_endpoint: "http://s3.invalid:9000".to_string(),
      public_endpoint_configured: false,
      cloud: true,
@@ -216,7 +67,7 @@ async fn a_launcher_syncs_what_a_session_changed() {
    "teams/t1/profiles/p1/manifest.json",
  ] {
    let out = service
-      .presign_upload(&body(json!({ "key": key })), &launcher)
+      .presign_upload(&body(json!({ "key": key })), &launcher, ORIGIN)
      .await
      .unwrap();
    assert!(out["url"].is_string());
@@ -225,6 +76,7 @@ async fn a_launcher_syncs_what_a_session_changed() {
    .presign_upload_batch(
      &body(json!({ "items": [{ "key": "profiles/p1/files/Default/Preferences" }] })),
      &launcher,
+      ORIGIN,
    )
    .await
    .unwrap();
@@ -257,7 +109,7 @@ async fn every_other_launcher_write_is_refused_before_it_reaches_s3() {
    assert_eq!(
      refusal_code(
        service
-          .presign_upload(&body(json!({ "key": key })), &launcher)
+          .presign_upload(&body(json!({ "key": key })), &launcher, ORIGIN)
          .await
      ),
      "TEAM_ROLE_READ_ONLY",
@@ -273,6 +125,7 @@ async fn every_other_launcher_write_is_refused_before_it_reaches_s3() {
            { "key": "profiles/p1/metadata.json" },
          ] })),
          &launcher,
+          ORIGIN,
        )
        .await
    ),
@@ -311,6 +164,7 @@ async fn a_member_changes_profile_settings_and_deletes_profiles() {
    .presign_upload(
      &body(json!({ "key": "profiles/p1/metadata.json" })),
      &member,
+      ORIGIN,
    )
    .await
    .unwrap();
@@ -402,7 +256,7 @@ async fn a_limited_member_sees_only_the_profiles_and_groups_it_may_reach() {
    );
  }
  let out = service
-    .presign_download(&body(json!({ "key": "proxies/x.json" })), &ctx)
+    .presign_download(&body(json!({ "key": "proxies/x.json" })), &ctx, ORIGIN)
    .await
    .unwrap();
  assert!(out["url"].is_string());
@@ -436,6 +290,7 @@ async fn a_limited_member_changes_profiles_only_inside_its_groups() {
    .presign_upload(
      &body(json!({ "key": "profiles/new/metadata.json", "metadata": { "group-id": "g1" } })),
      &ctx,
+      ORIGIN,
    )
    .await
    .unwrap();
@@ -443,6 +298,7 @@ async fn a_limited_member_changes_profiles_only_inside_its_groups() {
    .presign_upload(
      &body(json!({ "key": "profiles/p1/files/Default/Cookies" })),
      &ctx,
+      ORIGIN,
    )
    .await
    .unwrap();
@@ -455,7 +311,11 @@ async fn a_limited_member_changes_profiles_only_inside_its_groups() {
    json!({ "key": "groups/g2.json" }),
  ] {
    assert_eq!(
-      refusal_code(service.presign_upload(&body(request.clone()), &ctx).await),
+      refusal_code(
+        service
+          .presign_upload(&body(request.clone()), &ctx, ORIGIN)
+          .await
+      ),
      "TEAM_GROUP_ACCESS_DENIED",
      "{request}"
    );
@@ -667,12 +527,16 @@ async fn a_full_plan_keeps_syncing_its_profiles_and_settings_but_adds_none() {
    "groups/g.json",
  ] {
    service
-      .presign_upload(&body(json!({ "key": key })), &ctx)
+      .presign_upload(&body(json!({ "key": key })), &ctx, ORIGIN)
      .await
      .unwrap();
  }
  let error = service
-    .presign_upload(&body(json!({ "key": "profiles/p3/metadata.json" })), &ctx)
+    .presign_upload(
+      &body(json!({ "key": "profiles/p3/metadata.json" })),
+      &ctx,
+      ORIGIN,
+    )
    .await
    .unwrap_err();
  assert_eq!(error.status(), 403);
@@ -686,6 +550,7 @@ async fn a_full_plan_keeps_syncing_its_profiles_and_settings_but_adds_none() {
        json!({ "items": [{ "key": "profiles/p1/files/a" }, { "key": "profiles/p3/files/a" }] }),
      ),
      &ctx,
+      ORIGIN,
    )
    .await
    .unwrap_err();
@@ -702,6 +567,7 @@ async fn a_batch_counts_its_new_profiles_together() {
        json!({ "items": [{ "key": "profiles/p2/files/a" }, { "key": "profiles/p2/files/b" }] }),
      ),
      &ctx,
+      ORIGIN,
    )
    .await
    .unwrap();
@@ -711,6 +577,7 @@ async fn a_batch_counts_its_new_profiles_together() {
        json!({ "items": [{ "key": "profiles/p2/files/a" }, { "key": "profiles/p3/files/a" }] }),
      ),
      &ctx,
+      ORIGIN,
    )
    .await
    .unwrap_err();
@@ -723,20 +590,21 @@ async fn the_profile_limit_applies_in_the_cloud_only() {
  let ctx = cloud_with_limit();
  assert_eq!(
    service
-      .presign_upload(&body(json!({ "key": "profiles/p1/files/a" })), &ctx)
+      .presign_upload(&body(json!({ "key": "profiles/p1/files/a" })), &ctx, ORIGIN)
      .await
      .unwrap_err()
      .status(),
    403
  );
  service
-    .presign_upload(&body(json!({ "key": "proxies/x.json" })), &ctx)
+    .presign_upload(&body(json!({ "key": "proxies/x.json" })), &ctx, ORIGIN)
    .await
    .unwrap();
  service
    .presign_upload(
      &body(json!({ "key": "profiles/p9/files/a" })),
      &UserContext::self_hosted(),
+      ORIGIN,
    )
    .await
    .unwrap();
@@ -747,6 +615,7 @@ async fn the_profile_limit_applies_in_the_cloud_only() {
        mode: Mode::Account,
        ..ctx
      },
+      ORIGIN,
    )
    .await
    .unwrap();
@@ -807,6 +676,7 @@ async fn presigned_uploads_echo_only_the_metadata_sync_uses() {
        "expiresIn": 999999,
      })),
      &UserContext::self_hosted(),
+      ORIGIN,
    )
    .await
    .unwrap();
@@ -156,6 +156,10 @@ pub fn parse_http_date_ms(s: &str) -> Option<i64> {
    .map(|d| d.as_millis() as i64)
 }

+pub fn http_date(ms: i64) -> String {
+  httpdate::fmt_http_date(UNIX_EPOCH + std::time::Duration::from_millis(ms.max(0) as u64))
+}
+
 #[cfg(test)]
 mod tests {
  use super::*;
@@ -0,0 +1,395 @@
+//! Transfers through this server, for a store sealed at rest. Devices cannot
+//! go to storage directly, so the presign routes hand out URLs to `PATH` here,
+//! signed and expiring like S3's, and the body streams through the sealed
+//! store. A device sends and reads exactly what it did with S3: the same
+//! method, `Content-Length`, `Content-Type` and the signed `x-amz-meta-*`.
+
+use std::sync::Arc;
+
+use bytes::Bytes;
+use futures_util::StreamExt;
+use hmac::{Hmac, KeyInit, Mac};
+use http_body_util::{BodyExt, StreamBody};
+use hyper::body::{Frame, Incoming};
+use hyper::http::request::Parts;
+use hyper::{Method, Response, header};
+use sha2::Sha256;
+
+use crate::at_rest::Keyring;
+use crate::http::{BoxError, ResBody, drain, full};
+use crate::store::{Store, StoreError, Upload};
+use crate::time;
+
+pub const PATH: &str = "/v1/transfer";
+/// Transfers streamed at a time; more wait their turn. Each holds a few
+/// frames in memory.
+const MAX_TRANSFERS: usize = 64;
+const META_PARAM: &str = "meta-";
+
+pub struct TransferSigner {
+  /// The current key's first; any of them verifies.
+  keys: Vec<[u8; 32]>,
+  slots: Arc<tokio::sync::Semaphore>,
+}
+
+/// What a valid URL lets its holder do.
+#[derive(Debug, PartialEq)]
+pub struct Grant {
+  pub key: String,
+  pub metadata: Vec<(String, String)>,
+}
+
+#[derive(Debug, PartialEq)]
+pub enum Refusal {
+  Expired,
+  BadSignature,
+}
+
+fn field(mac: &mut Hmac<Sha256>, value: &str) {
+  mac.update(&(value.len() as u64).to_be_bytes());
+  mac.update(value.as_bytes());
+}
+
+impl TransferSigner {
+  pub fn new(keyring: &Keyring) -> Self {
+    Self {
+      keys: keyring.url_keys(),
+      slots: Arc::new(tokio::sync::Semaphore::new(MAX_TRANSFERS)),
+    }
+  }
+
+  fn mac(
+    key: &[u8; 32],
+    method: &str,
+    object_key: &str,
+    expires_ms: &str,
+    metadata: &[(String, String)],
+  ) -> Hmac<Sha256> {
+    let mut mac = Hmac::<Sha256>::new_from_slice(key).expect("HMAC takes any key length");
+    for value in ["v1", method, object_key, expires_ms] {
+      field(&mut mac, value);
+    }
+    let mut sorted = metadata.to_vec();
+    sorted.sort();
+    for (name, value) in &sorted {
+      field(&mut mac, name);
+      field(&mut mac, value);
+    }
+    mac
+  }
+
+  /// `origin` is how devices reach this server, e.g. `https://sync.example.com`.
+  pub fn url(
+    &self,
+    origin: &str,
+    method: &str,
+    object_key: &str,
+    expires_ms: i64,
+    metadata: &[(String, String)],
+  ) -> String {
+    let expires = expires_ms.to_string();
+    let signature = Self::mac(&self.keys[0], method, object_key, &expires, metadata)
+      .finalize()
+      .into_bytes();
+    let mut query = url::form_urlencoded::Serializer::new(String::new());
+    query.append_pair("key", object_key);
+    query.append_pair("expires", &expires);
+    for (name, value) in metadata {
+      query.append_pair(&format!("{META_PARAM}{name}"), value);
+    }
+    query.append_pair("sig", &hex::encode(signature));
+    format!("{origin}{PATH}?{}", query.finish())
+  }
+
+  pub fn verify(&self, method: &str, query: &str, now_ms: i64) -> Result<Grant, Refusal> {
+    let mut key = None;
+    let mut expires = None;
+    let mut signature = None;
+    let mut metadata = Vec::new();
+    for (name, value) in url::form_urlencoded::parse(query.as_bytes()) {
+      match name.as_ref() {
+        "key" if key.is_none() => key = Some(value.into_owned()),
+        "expires" if expires.is_none() => expires = Some(value.into_owned()),
+        "sig" if signature.is_none() => signature = Some(value.into_owned()),
+        other => match other.strip_prefix(META_PARAM) {
+          Some(meta) => metadata.push((meta.to_string(), value.into_owned())),
+          None => return Err(Refusal::BadSignature),
+        },
+      }
+    }
+    let (Some(key), Some(expires), Some(signature)) = (key, expires, signature) else {
+      return Err(Refusal::BadSignature);
+    };
+    let signature = hex::decode(signature).map_err(|_| Refusal::BadSignature)?;
+    let signed = self.keys.iter().any(|k| {
+      Self::mac(k, method, &key, &expires, &metadata)
+        .verify_slice(&signature)
+        .is_ok()
+    });
+    if !signed {
+      return Err(Refusal::BadSignature);
+    }
+    let expires_ms: i64 = expires.parse().map_err(|_| Refusal::BadSignature)?;
+    if now_ms > expires_ms {
+      return Err(Refusal::Expired);
+    }
+    Ok(Grant { key, metadata })
+  }
+}
+
+/// An S3-style error document, so a device logs what S3 would have said.
+fn s3_error(status: u16, code: &str, message: &str) -> Response<ResBody> {
+  let body = format!(
+    "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n<Error><Code>{code}</Code><Message>{message}</Message></Error>"
+  );
+  Response::builder()
+    .status(status)
+    .header(header::CONTENT_TYPE, "application/xml")
+    .header(header::CONTENT_LENGTH, body.len())
+    .body(full(body))
+    .expect("static response parts are valid")
+}
+
+fn store_refusal(error: StoreError) -> Response<ResBody> {
+  match error {
+    StoreError::NotFound => s3_error(404, "NoSuchKey", "The specified key does not exist."),
+    StoreError::Service { status, code, .. } if (400..500).contains(&status) => {
+      s3_error(status, &code, "")
+    }
+    other => {
+      crate::log::error(PATH, &format!("Transfer failed err=\"{other}\""));
+      s3_error(
+        500,
+        "InternalError",
+        "We encountered an internal error. Please try again.",
+      )
+    }
+  }
+}
+
+fn header_text(parts: &Parts, name: impl header::AsHeaderName) -> Option<String> {
+  parts
+    .headers
+    .get(name)
+    .map(|v| v.as_bytes().iter().map(|&b| b as char).collect())
+}
+
+pub async fn serve(
+  store: &Arc<dyn Store>,
+  signer: &TransferSigner,
+  parts: &Parts,
+  body: Incoming,
+) -> Response<ResBody> {
+  let method = parts.method.as_str();
+  let grant = match signer.verify(method, parts.uri.query().unwrap_or(""), time::now_ms()) {
+    Ok(grant) if parts.method == Method::GET || parts.method == Method::PUT => grant,
+    Ok(_) | Err(Refusal::BadSignature) => {
+      drain(body).await;
+      return s3_error(
+        403,
+        "SignatureDoesNotMatch",
+        "The request signature we calculated does not match the signature you provided.",
+      );
+    }
+    Err(Refusal::Expired) => {
+      drain(body).await;
+      return s3_error(403, "AccessDenied", "Request has expired");
+    }
+  };
+  let slot = signer
+    .slots
+    .clone()
+    .acquire_owned()
+    .await
+    .expect("the semaphore is never closed");
+  if parts.method == Method::GET {
+    drain(body).await;
+    return download(store, &grant.key, slot).await;
+  }
+  let response = upload(store, grant, parts, body).await;
+  drop(slot);
+  response
+}
+
+async fn download(
+  store: &Arc<dyn Store>,
+  key: &str,
+  slot: tokio::sync::OwnedSemaphorePermit,
+) -> Response<ResBody> {
+  let object = match store.get_object(key).await {
+    Ok(object) => object,
+    Err(e) => return store_refusal(e),
+  };
+  let mut response = Response::builder().status(200);
+  if let Some(length) = object.head.content_length {
+    response = response.header(header::CONTENT_LENGTH, length);
+  }
+  if let Some(content_type) = &object.head.content_type {
+    response = response.header(header::CONTENT_TYPE, content_type.as_str());
+  }
+  if let Some(ms) = object.head.last_modified_ms {
+    response = response.header(header::LAST_MODIFIED, time::http_date(ms));
+  }
+  for (name, value) in &object.head.metadata {
+    response = response.header(format!("x-amz-meta-{name}"), value.as_str());
+  }
+  let key = key.to_string();
+  // The slot is held until the body is sent.
+  let body = object.body.map(move |chunk| {
+    let _slot = &slot;
+    chunk.map(Frame::data).map_err(|e| {
+      crate::log::error(PATH, &format!("Download cut off key={key} err=\"{e}\""));
+      BoxError::from(e)
+    })
+  });
+  response
+    .body(StreamBody::new(body).boxed_unsync())
+    .unwrap_or_else(|_| {
+      s3_error(
+        500,
+        "InternalError",
+        "Stored metadata is not a header value.",
+      )
+    })
+}
+
+async fn upload(
+  store: &Arc<dyn Store>,
+  grant: Grant,
+  parts: &Parts,
+  body: Incoming,
+) -> Response<ResBody> {
+  let mut sent: Vec<(String, String)> = parts
+    .headers
+    .iter()
+    .filter_map(|(name, value)| {
+      name.as_str().strip_prefix("x-amz-meta-").map(|n| {
+        (
+          n.to_string(),
+          value.as_bytes().iter().map(|&b| b as char).collect(),
+        )
+      })
+    })
+    .collect();
+  sent.sort();
+  let mut signed = grant.metadata;
+  signed.sort();
+  if sent != signed {
+    drain(body).await;
+    return s3_error(
+      403,
+      "AccessDenied",
+      "There were headers present in the request which were not signed",
+    );
+  }
+  let Some(length) = header_text(parts, header::CONTENT_LENGTH).and_then(|v| v.trim().parse().ok())
+  else {
+    drain(body).await;
+    return s3_error(
+      411,
+      "MissingContentLength",
+      "You must provide the Content-Length HTTP header.",
+    );
+  };
+  let content_type = header_text(parts, header::CONTENT_TYPE);
+  let stream = body
+    .into_data_stream()
+    .map(|chunk: Result<Bytes, hyper::Error>| {
+      chunk.map_err(|e| StoreError::Transport(e.to_string()))
+    });
+  match store
+    .put_object_stream(
+      &grant.key,
+      Box::pin(stream),
+      Upload {
+        length,
+        content_type: content_type.as_deref(),
+        metadata: &signed,
+        if_match: None,
+      },
+    )
+    .await
+  {
+    Ok(()) => Response::builder()
+      .status(200)
+      .header(header::CONTENT_LENGTH, 0)
+      .body(full(Bytes::new()))
+      .expect("static response parts are valid"),
+    Err(e) => store_refusal(e),
+  }
+}
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+
+  fn signer() -> TransferSigner {
+    TransferSigner::new(&Keyring::parse("k1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=").unwrap())
+  }
+
+  fn query(url: &str) -> &str {
+    url.split_once('?').unwrap().1
+  }
+
+  #[test]
+  fn urls_verify_for_their_method_key_metadata_and_lifetime() {
+    let signer = signer();
+    let metadata = vec![
+      ("updated-at".to_string(), "17".to_string()),
+      ("group-id".to_string(), "g 1&x".to_string()),
+    ];
+    let url = signer.url(
+      "https://sync.example.com",
+      "PUT",
+      "users/u/a b+c",
+      1000,
+      &metadata,
+    );
+    assert!(url.starts_with("https://sync.example.com/v1/transfer?key=users%2Fu%2Fa+b%2Bc&"));
+    let grant = signer.verify("PUT", query(&url), 1000).unwrap();
+    assert_eq!(grant.key, "users/u/a b+c");
+    assert_eq!(grant.metadata.len(), 2);
+    assert_eq!(
+      signer.verify("PUT", query(&url), 1001),
+      Err(Refusal::Expired)
+    );
+    assert_eq!(
+      signer.verify("GET", query(&url), 0),
+      Err(Refusal::BadSignature)
+    );
+    let tampered = query(&url).replace("updated-at=17", "updated-at=18");
+    assert_eq!(
+      signer.verify("PUT", &tampered, 0),
+      Err(Refusal::BadSignature)
+    );
+    let moved = query(&url).replace("key=users%2Fu", "key=users%2Fv");
+    assert_eq!(signer.verify("PUT", &moved, 0), Err(Refusal::BadSignature));
+    let added = format!("{}&meta-x=1", query(&url));
+    assert_eq!(signer.verify("PUT", &added, 0), Err(Refusal::BadSignature));
+    let extended = query(&url).replace("expires=1000", "expires=9999");
+    assert_eq!(
+      signer.verify("PUT", &extended, 0),
+      Err(Refusal::BadSignature)
+    );
+  }
+
+  #[test]
+  fn an_old_key_still_verifies_after_rotation() {
+    let old = signer();
+    let url = old.url("http://h", "GET", "k", 50, &[]);
+    let rotated = TransferSigner::new(
+      &Keyring::parse(
+        "k2:AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE=,k1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=",
+      )
+      .unwrap(),
+    );
+    assert!(rotated.verify("GET", query(&url), 0).is_ok());
+    let removed = TransferSigner::new(
+      &Keyring::parse("k2:AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE=").unwrap(),
+    );
+    assert_eq!(
+      removed.verify("GET", query(&url), 0),
+      Err(Refusal::BadSignature)
+    );
+  }
+}
@@ -0,0 +1,452 @@
+//! Encryption at rest end to end: the binary with `SYNC_AT_REST_KEY` against
+//! the compose S3 server, devices that speak only the S3 presigned protocol,
+//! plaintext objects from before, and the migration command.
+
+mod common;
+
+use common::{SYNC_TOKEN, Server, bucket, http, post, raw_store, run_command, storage_env};
+use donut_sync::store::{Store, Upload, collect, once};
+use serde_json::{Value, json};
+
+const K1: &str = "k1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=";
+const K2: &str = "k2:AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQE=";
+
+fn sample(len: usize) -> Vec<u8> {
+  (0..len).map(|i| (i * 131 + 17) as u8).collect()
+}
+
+async fn sealed_server(name: &str, keys: &str) -> (Server, String) {
+  let bucket = bucket(name);
+  (
+    Server::start(&bucket, &[("SYNC_AT_REST_KEY", Some(keys))]).await,
+    bucket,
+  )
+}
+
+async fn call(server: &Server, route: &str, body: Value) -> (u16, Value) {
+  post(
+    &http(),
+    &format!("{}/v1/objects/{route}", server.url),
+    Some(SYNC_TOKEN),
+    body,
+  )
+  .await
+}
+
+/// What the app does: presign, then PUT with the echoed metadata.
+async fn upload(server: &Server, key: &str, body: Vec<u8>, updated_at: Option<&str>) -> String {
+  let mut request = json!({ "key": key, "contentType": "application/octet-stream" });
+  if let Some(updated_at) = updated_at {
+    request["metadata"] = json!({ "updated-at": updated_at });
+  }
+  let (status, presign) = call(server, "presign-upload", request).await;
+  assert_eq!(status, 200, "{presign}");
+  let url = presign["url"].as_str().unwrap().to_string();
+  let mut put = http()
+    .put(&url)
+    .header("Content-Length", body.len().to_string())
+    .header("Content-Type", "application/octet-stream")
+    .body(body);
+  if let Some(metadata) = presign["metadata"].as_object() {
+    for (name, value) in metadata {
+      put = put.header(format!("x-amz-meta-{name}"), value.as_str().unwrap());
+    }
+  }
+  let response = put.send().await.unwrap();
+  assert_eq!(response.status(), 200, "{}", response.text().await.unwrap());
+  url
+}
+
+async fn download(server: &Server, key: &str) -> (u16, Vec<u8>) {
+  let (status, presign) = call(server, "presign-download", json!({ "key": key })).await;
+  assert_eq!(status, 200, "{presign}");
+  let response = http()
+    .get(presign["url"].as_str().unwrap())
+    .send()
+    .await
+    .unwrap();
+  let status = response.status().as_u16();
+  (status, response.bytes().await.unwrap().to_vec())
+}
+
+#[tokio::test]
+async fn transfers_go_through_the_server_and_are_sealed_in_storage() {
+  let (server, bucket) = sealed_server("seal", K1).await;
+  let raw = raw_store(&bucket);
+  for (key, size) in [
+    ("profiles/p/files/big", 3 * 65_536 + 77),
+    ("profiles/p/files/frame", 65_536),
+    ("profiles/p/files/empty", 0),
+  ] {
+    let plain = sample(size);
+    let url = upload(&server, key, plain.clone(), Some("1700000000")).await;
+    assert!(
+      url.starts_with(&format!("{}/v1/transfer?", server.url)),
+      "{url}"
+    );
+
+    let stored = collect(raw.get_object(key).await.unwrap().body)
+      .await
+      .unwrap();
+    assert_eq!(&stored[..4], b"DSAR");
+    assert_ne!(stored, plain);
+    let head = raw.head_object(key).await.unwrap();
+    assert_eq!(head.metadata_value("donut-at-rest"), Some("1:k1"));
+
+    let (status, stat) = call(&server, "stat", json!({ "key": key })).await;
+    assert_eq!(status, 200);
+    assert_eq!(stat["size"], json!(size));
+    assert_eq!(stat["metadata"], json!({ "updated-at": "1700000000" }));
+
+    assert_eq!(download(&server, key).await, (200, plain));
+  }
+  let (_, readyz) = (
+    0,
+    http()
+      .get(format!("{}/readyz", server.url))
+      .send()
+      .await
+      .unwrap()
+      .json::<Value>()
+      .await
+      .unwrap(),
+  );
+  assert_eq!(readyz, json!({ "status": "ready", "s3": true }));
+  assert!(
+    server
+      .logs()
+      .contains("Encryption at rest on key=k1 keys=1")
+  );
+}
+
+#[tokio::test]
+async fn batches_and_downloads_keep_the_s3_shapes() {
+  let (server, _) = sealed_server("batch", K1).await;
+  let (status, batch) = call(
+    &server,
+    "presign-upload-batch",
+    json!({ "items": [{ "key": "a" }, { "key": "b" }] }),
+  )
+  .await;
+  assert_eq!(status, 200);
+  for item in batch["items"].as_array().unwrap() {
+    let key = item["key"].as_str().unwrap();
+    assert!(item["expiresAt"].is_string());
+    let response = http()
+      .put(item["url"].as_str().unwrap())
+      .header("Content-Length", key.len().to_string())
+      .body(key.to_string())
+      .send()
+      .await
+      .unwrap();
+    assert_eq!(response.status(), 200);
+  }
+  let (_, batch) = call(
+    &server,
+    "presign-download-batch",
+    json!({ "keys": ["a", "b"] }),
+  )
+  .await;
+  for item in batch["items"].as_array().unwrap() {
+    let response = http()
+      .get(item["url"].as_str().unwrap())
+      .send()
+      .await
+      .unwrap();
+    assert_eq!(
+      response.text().await.unwrap(),
+      item["key"].as_str().unwrap()
+    );
+  }
+  let (_, list) = call(&server, "list", json!({ "prefix": "" })).await;
+  let keys: Vec<&str> = list["objects"]
+    .as_array()
+    .unwrap()
+    .iter()
+    .map(|o| o["key"].as_str().unwrap())
+    .collect();
+  assert_eq!(keys, ["a", "b"]);
+}
+
+#[tokio::test]
+async fn transfer_urls_refuse_what_s3_would_refuse() {
+  let (server, _) = sealed_server("refuse", K1).await;
+  let client = http();
+  let (_, presign) = call(
+    &server,
+    "presign-upload",
+    json!({ "key": "k", "metadata": { "updated-at": "5" } }),
+  )
+  .await;
+  let url = presign["url"].as_str().unwrap();
+  let put = |url: String| client.put(url).header("Content-Length", "1").body("x");
+
+  // The signed metadata must be sent, exactly.
+  let response = put(url.to_string()).send().await.unwrap();
+  assert_eq!(response.status(), 403);
+  let response = put(url.to_string())
+    .header("x-amz-meta-updated-at", "6")
+    .send()
+    .await
+    .unwrap();
+  assert_eq!(response.status(), 403);
+  let response = put(url.to_string())
+    .header("x-amz-meta-updated-at", "5")
+    .header("x-amz-meta-group-id", "g")
+    .send()
+    .await
+    .unwrap();
+  assert_eq!(response.status(), 403);
+  // A changed URL, or the wrong method.
+  let response = put(url.replace("updated-at=5", "updated-at=6"))
+    .header("x-amz-meta-updated-at", "6")
+    .send()
+    .await
+    .unwrap();
+  assert_eq!(response.status(), 403);
+  assert!(
+    response
+      .text()
+      .await
+      .unwrap()
+      .contains("SignatureDoesNotMatch")
+  );
+  let response = client.get(url).send().await.unwrap();
+  assert_eq!(response.status(), 403);
+  // No length, as S3 answers a chunked PUT.
+  let chunked = reqwest::Body::wrap_stream(futures_util::stream::iter([Ok::<_, std::io::Error>(
+    bytes::Bytes::from_static(b"x"),
+  )]));
+  let response = client
+    .put(url)
+    .header("x-amz-meta-updated-at", "5")
+    .body(chunked)
+    .send()
+    .await
+    .unwrap();
+  assert_eq!(response.status(), 411);
+  // Expired.
+  let (_, short) = call(
+    &server,
+    "presign-download",
+    json!({ "key": "k", "expiresIn": 0.001 }),
+  )
+  .await;
+  tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+  let response = client
+    .get(short["url"].as_str().unwrap())
+    .send()
+    .await
+    .unwrap();
+  assert_eq!(response.status(), 403);
+  assert!(
+    response
+      .text()
+      .await
+      .unwrap()
+      .contains("Request has expired")
+  );
+  // Missing object.
+  assert_eq!(download(&server, "missing").await.0, 404);
+  // The right request still works.
+  let response = put(url.to_string())
+    .header("x-amz-meta-updated-at", "5")
+    .send()
+    .await
+    .unwrap();
+  assert_eq!(response.status(), 200);
+}
+
+#[tokio::test]
+async fn without_the_key_nothing_changes() {
+  let server = Server::start(&bucket("plain"), &[]).await;
+  let (_, presign) = call(&server, "presign-upload", json!({ "key": "k" })).await;
+  assert!(
+    presign["url"]
+      .as_str()
+      .unwrap()
+      .starts_with("http://127.0.0.1:8987/")
+  );
+  let response = http()
+    .get(format!("{}/v1/transfer?key=k", server.url))
+    .send()
+    .await
+    .unwrap();
+  assert_eq!(response.status(), 404);
+  assert_eq!(
+    response.json::<Value>().await.unwrap()["message"],
+    "Cannot GET /v1/transfer?key=k"
+  );
+  assert!(!server.logs().contains("Encryption at rest"));
+}
+
+/// Writes `objects` straight to storage, as an older server left them.
+async fn seed_plaintext(bucket: &str, objects: &[(&str, Vec<u8>)]) {
+  let raw = raw_store(bucket);
+  raw.create_bucket().await.ok();
+  for (key, body) in objects {
+    raw
+      .put_object_stream(
+        key,
+        once(body.clone()),
+        Upload {
+          length: body.len() as u64,
+          content_type: Some("application/json"),
+          metadata: &[("updated-at".to_string(), "9".to_string())],
+          if_match: None,
+        },
+      )
+      .await
+      .unwrap();
+  }
+}
+
+#[tokio::test]
+async fn plaintext_and_sealed_objects_read_side_by_side() {
+  let bucket = bucket("mixed");
+  let old = sample(100_000);
+  seed_plaintext(&bucket, &[("proxies/old.json", old.clone())]).await;
+  let server = Server::start(&bucket, &[("SYNC_AT_REST_KEY", Some(K1))]).await;
+  upload(&server, "proxies/new.json", b"new".to_vec(), Some("10")).await;
+
+  assert_eq!(download(&server, "proxies/old.json").await, (200, old));
+  assert_eq!(
+    download(&server, "proxies/new.json").await,
+    (200, b"new".to_vec())
+  );
+  let (_, stat) = call(&server, "stat", json!({ "key": "proxies/old.json" })).await;
+  assert_eq!(stat["size"], json!(100_000));
+  assert_eq!(stat["metadata"], json!({ "updated-at": "9" }));
+
+  // Restore points copy sealed objects and read their sealed records.
+  upload(&server, "profiles/p/manifest.json", b"{}".to_vec(), None).await;
+  upload(&server, "profiles/p/files/a", b"A".to_vec(), None).await;
+  let (status, snapshot) = call(
+    &server,
+    "snapshots/create",
+    json!({ "profileRoot": "profiles/p/" }),
+  )
+  .await;
+  assert_eq!(status, 200, "{snapshot}");
+  let (_, listed) = call(
+    &server,
+    "snapshots/list",
+    json!({ "profileRoot": "profiles/p/" }),
+  )
+  .await;
+  assert_eq!(listed["snapshots"][0]["id"], snapshot["id"]);
+  upload(&server, "profiles/p/files/a", b"changed".to_vec(), None).await;
+  let (status, _) = call(
+    &server,
+    "snapshots/restore-files",
+    json!({ "profileRoot": "profiles/p/", "id": snapshot["id"] }),
+  )
+  .await;
+  assert_eq!(status, 200);
+  assert_eq!(
+    download(&server, "profiles/p/files/a").await,
+    (200, b"A".to_vec())
+  );
+}
+
+#[tokio::test]
+async fn migration_seals_everything_in_place_and_a_rerun_does_nothing() {
+  let bucket = bucket("migrate");
+  let objects: Vec<(String, Vec<u8>)> = (0..25)
+    .map(|i| {
+      (
+        format!("users/u/profiles/p/files/{i:02}"),
+        sample(i * 9_000),
+      )
+    })
+    .collect();
+  let refs: Vec<(&str, Vec<u8>)> = objects
+    .iter()
+    .map(|(k, v)| (k.as_str(), v.clone()))
+    .collect();
+  seed_plaintext(&bucket, &refs).await;
+  let storage = storage_env(&bucket);
+  let mut env: Vec<(&str, &str)> = storage.iter().map(|(k, v)| (*k, v.as_str())).collect();
+  env.push(("SYNC_AT_REST_KEY", K1));
+  let total: usize = objects.iter().map(|(_, v)| v.len()).sum();
+
+  let (code, out) = run_command(&["migrate-at-rest", "--dry-run"], &env);
+  assert_eq!(code, 0, "{out}");
+  assert!(
+    out.contains(&format!(
+      "scanned=25 current=0 sealed=25 sealed_bytes={total} "
+    )),
+    "{out}"
+  );
+  let raw = raw_store(&bucket);
+  let first = collect(raw.get_object(&objects[3].0).await.unwrap().body)
+    .await
+    .unwrap();
+  assert_eq!(first, objects[3].1);
+
+  let (code, out) = run_command(
+    &[
+      "migrate-at-rest",
+      "--concurrency",
+      "3",
+      "--max-mib-per-sec",
+      "64",
+    ],
+    &env,
+  );
+  assert_eq!(code, 0, "{out}");
+  assert!(out.contains("conditional_put=false"), "{out}");
+  assert!(out.contains("sealed=25"), "{out}");
+  for (key, _) in &objects {
+    let head = raw.head_object(key).await.unwrap();
+    assert_eq!(head.metadata_value("donut-at-rest"), Some("1:k1"), "{key}");
+    assert_eq!(head.metadata_value("updated-at"), Some("9"), "{key}");
+    assert_eq!(head.content_type.as_deref(), Some("application/json"));
+  }
+  let listed = raw
+    .list_objects_v2(&donut_sync::store::ListInput::default())
+    .await
+    .unwrap();
+  assert_eq!(listed.contents.len(), 25, "no temporary copy is left");
+
+  let (code, out) = run_command(&["migrate-at-rest"], &env);
+  assert_eq!(code, 0, "{out}");
+  assert!(out.contains("current=25 sealed=0"), "{out}");
+
+  // The server reads what the migration wrote. Self-hosted keys are bare.
+  let server = Server::start(&bucket, &[("SYNC_AT_REST_KEY", Some(K1))]).await;
+  for (key, body) in &objects {
+    assert_eq!(download(&server, key).await, (200, body.clone()));
+  }
+  drop(server);
+
+  // Rotation: k2 writes, k1 still reads, and the migration moves k1's objects.
+  let rotated = format!("{K2},{K1}");
+  let server = Server::start(&bucket, &[("SYNC_AT_REST_KEY", Some(&rotated))]).await;
+  assert_eq!(
+    download(&server, &objects[5].0).await,
+    (200, objects[5].1.clone())
+  );
+  drop(server);
+  env.retain(|(k, _)| *k != "SYNC_AT_REST_KEY");
+  env.push(("SYNC_AT_REST_KEY", &rotated));
+  let (code, out) = run_command(&["migrate-at-rest"], &env);
+  assert_eq!(code, 0, "{out}");
+  assert!(out.contains("rotated=25"), "{out}");
+  let server = Server::start(&bucket, &[("SYNC_AT_REST_KEY", Some(K2))]).await;
+  assert_eq!(
+    download(&server, &objects[7].0).await,
+    (200, objects[7].1.clone())
+  );
+}
+
+#[tokio::test]
+async fn a_bad_key_stops_the_server_at_start() {
+  let storage = storage_env(&bucket("badkey"));
+  let mut env: Vec<(&str, &str)> = storage.iter().map(|(k, v)| (*k, v.as_str())).collect();
+  env.push(("SYNC_TOKEN", SYNC_TOKEN));
+  env.push(("SYNC_AT_REST_KEY", "k1:AAAA"));
+  let (code, out) = common::run_to_exit(&env);
+  assert_eq!(code, 1);
+  assert!(out.contains("must be 32 bytes"), "{out}");
+}
@@ -15,6 +15,29 @@ use donut_sync::store::{Store, StoreError};
 pub const S3_ENDPOINT: &str = "http://127.0.0.1:8987";
 pub const SYNC_TOKEN: &str = "test-sync-token-0123456789abcdef";

+/// The test storage, without the server: what is really stored.
+pub fn raw_store(bucket: &str) -> S3Store {
+  S3Store::new(
+    Location::parse(S3_ENDPOINT, bucket, true).unwrap(),
+    Credentials {
+      access_key_id: "minioadmin".to_string(),
+      secret_access_key: "minioadmin".to_string(),
+    },
+    "us-east-1".to_string(),
+  )
+}
+
+/// The storage settings the server takes, for a command run to its exit.
+pub fn storage_env(bucket: &str) -> Vec<(&'static str, String)> {
+  vec![
+    ("S3_ENDPOINT", S3_ENDPOINT.to_string()),
+    ("S3_ACCESS_KEY_ID", "minioadmin".to_string()),
+    ("S3_SECRET_ACCESS_KEY", "minioadmin".to_string()),
+    ("S3_BUCKET", bucket.to_string()),
+    ("S3_FORCE_PATH_STYLE", "true".to_string()),
+  ]
+}
+
 pub fn http() -> reqwest::Client {
  let _ = rustls::crypto::ring::default_provider().install_default();
  reqwest::Client::builder().no_proxy().build().unwrap()
@@ -78,7 +101,7 @@ impl Server {
      }
    }
    let output = Arc::new(Mutex::new(String::new()));
-    let mut child = spawn(&env);
+    let mut child = spawn(&[], &env);
    collect(&mut child, &output);
    let url = format!("http://127.0.0.1:{port}");
    let client = http();
@@ -114,11 +137,12 @@ impl Drop for Server {
  }
 }

-fn spawn(env: &[(String, String)]) -> Child {
+fn spawn(args: &[&str], env: &[(String, String)]) -> Child {
  // A directory without a `.env`, so a developer's file cannot leak in.
  let dir = std::env::temp_dir().join("donut-sync-test-cwd");
  std::fs::create_dir_all(&dir).unwrap();
  Command::new(env!("CARGO_BIN_EXE_donut-sync"))
+    .args(args)
    .env_clear()
    .envs(env.iter().map(|(k, v)| (k.as_str(), v.as_str())))
    .current_dir(dir)
@@ -149,11 +173,16 @@ fn collect(child: &mut Child, output: &Arc<Mutex<String>>) {
 /// Runs the binary to its exit (a refused configuration) and returns the
 /// exit code and what it printed.
 pub fn run_to_exit(env: &[(&str, &str)]) -> (i32, String) {
+  run_command(&[], env)
+}
+
+/// Runs the binary with `args` to its exit.
+pub fn run_command(args: &[&str], env: &[(&str, &str)]) -> (i32, String) {
  let env: Vec<(String, String)> = env
    .iter()
    .map(|(k, v)| (k.to_string(), v.to_string()))
    .collect();
-  let child = spawn(&env);
+  let child = spawn(args, &env);
  let out = child.wait_with_output().unwrap();
  (
    out.status.code().unwrap_or(-1),