Skip to main content

max / synckit

20.0 KB · 513 lines History Blame Raw
1 //! Master-key rotation orchestration.
2 //!
3 //! Drives the multi-round server protocol that re-encrypts every device's key
4 //! envelope under a new master key, completes the rotation, and sweeps up
5 //! stragglers. The round count is bounded by `MAX_ROTATION_ROUNDS` so a
6 //! misbehaving server cannot spin the loop indefinitely.
7
8 use bytes::Bytes;
9 use tracing::instrument;
10 use uuid::Uuid;
11
12 use crate::{
13 crypto,
14 error::Result,
15 ids::DeviceId,
16 types::{
17 BeginRotationRequest, BeginRotationResponse, CompleteRotationRequest, PendingKeyInfo,
18 RotationBatchEntry, RotationBatchRequest, RotationBatchResponse, RotationEntriesRequest,
19 RotationEntriesResponse,
20 },
21 };
22
23 use super::SyncKitClient;
24 use super::helpers::{Idempotency, check_response};
25
26 /// Hard ceiling on re-encrypt / straggler rounds within a single rotation. Each
27 /// round re-encrypts one server batch and the set provably shrinks (re-encrypted
28 /// entries drop out of the server's `key_id != new_key_id` filter), so a real
29 /// rotation finishes in `total_entries / batch_size` rounds. The cap only fires
30 /// when the set does NOT shrink, a stalled or hostile server that keeps handing
31 /// back work, converting an unbounded spin into a bounded error. Generous enough
32 /// to cover any real sync log at any batch size.
33 const MAX_ROTATION_ROUNDS: u32 = 100_000;
34
35 impl SyncKitClient {
36 /// Rotate the master encryption key.
37 ///
38 /// Generates a new 256-bit key, re-encrypts all sync log entries in batches,
39 /// and commits the new key to the server.
40 ///
41 /// Genuinely idempotent: if a previous call was interrupted, the server still
42 /// holds the rotation's `pending_key`. This call detects that and **resumes
43 /// with the already-committed key** rather than minting a second new key,
44 /// the latter would orphan any entries already re-encrypted under the first
45 /// key. Only when there is no pending rotation is a fresh key generated.
46 ///
47 /// Requires the encryption password (to wrap the new key) and a device_id
48 /// (the device performing the rotation). Only one device can rotate at a time.
49 ///
50 /// After completion, all devices will receive the new key on their next
51 /// `GET /keys` call. During rotation, both old and new keys are available
52 /// so pulls continue to work.
53 #[instrument(skip(self, password))]
54 pub async fn rotate_key(&self, device_id: DeviceId, password: &str) -> Result<()> {
55 // 1. Verify password against the still-active key, and learn whether a
56 // rotation is already pending for this user.
57 let key_state = self.get_server_key_full().await?;
58 let key_version = key_state.key_version.unwrap_or(0);
59 let old_key = crypto::verify_password_against_envelope(&key_state.encrypted_key, password)?;
60 let old_key = crypto::ZeroizeOnDrop(old_key);
61
62 // 2. Resume an interrupted rotation by adopting its committed pending key,
63 // or start fresh. Either way `new_envelope` is the key the server will
64 // promote on completion, so re-encryption uses the same key it commits.
65 let (new_master_key, new_envelope) =
66 Self::rotation_key_material(key_state.pending_key, password)?;
67 // Wrap immediately so the new key is zeroized on every exit path, an
68 // error in the re-encrypt or complete loops below would otherwise drop
69 // the bare array without scrubbing it.
70 let new_master_key = crypto::ZeroizeOnDrop(new_master_key);
71
72 // 3. Begin (or resume) rotation. `begin_key_rotation` is idempotent for the
73 // same device: it returns the existing rotation, ignoring `new_envelope`
74 // on resume, which is why step 2 must adopt the committed key.
75 let begin_resp = self
76 .begin_rotation(device_id, &new_envelope, key_version)
77 .await?;
78 let rotation_id = begin_resp.rotation_id;
79 let new_key_id = begin_resp.new_key_id;
80
81 tracing::info!(
82 rotation_id = %rotation_id,
83 target_seq = begin_resp.target_seq,
84 new_key_id = new_key_id,
85 "Key rotation started",
86 );
87
88 // 4. Re-encrypt loop (bounded, see reencrypt_until_done).
89 self.reencrypt_until_done(rotation_id, &old_key, &new_master_key)
90 .await?;
91
92 // 5. Complete, retry if stragglers arrived from concurrent pushes. Both
93 // the straggler retries and each re-encrypt pass are capped so a server
94 // that keeps returning 409 (a stall, or a hostile peer racing pushes)
95 // cannot spin this loop forever.
96 let mut straggler_rounds = 0u32;
97 loop {
98 match self.complete_rotation(rotation_id).await {
99 Ok(()) => break,
100 Err(crate::error::SyncKitError::Server { status: 409, .. }) => {
101 straggler_rounds += 1;
102 if straggler_rounds >= MAX_ROTATION_ROUNDS {
103 return Err(crate::error::SyncKitError::Internal(
104 "key rotation did not converge: server kept reporting stragglers past the round cap".into(),
105 ));
106 }
107 tracing::debug!(
108 round = straggler_rounds,
109 "Stragglers detected, re-encrypting remaining entries"
110 );
111 self.reencrypt_until_done(rotation_id, &old_key, &new_master_key)
112 .await?;
113 }
114 Err(e) => return Err(e),
115 }
116 }
117
118 // 6. Update local state
119 let (app_id, user_id) = self.require_session_ids()?;
120 crate::keystore::store_key(app_id, user_id, &new_master_key)?;
121 *self.master_key.write() = Some(new_master_key);
122 *self.master_key_id.write() = new_key_id;
123 *self.pending_key.write() = None;
124
125 tracing::info!("Key rotation completed");
126 Ok(())
127 }
128
129 /// Decide the rotation's new key material: adopt the server's committed
130 /// `pending_key` when resuming an interrupted rotation, otherwise mint a
131 /// fresh key. Returns `(new_master_key, new_envelope)` where the envelope is
132 /// what the server promotes on completion, so the re-encryption key always
133 /// matches the committed envelope. Pure (no IO) so it is unit-testable.
134 fn rotation_key_material(
135 pending: Option<PendingKeyInfo>,
136 password: &str,
137 ) -> Result<([u8; 32], String)> {
138 match pending {
139 Some(pending) => {
140 let key = crypto::unwrap_master_key(&pending.encrypted_key, password)?;
141 Ok((key, pending.encrypted_key))
142 }
143 None => {
144 // Wrap before the fallible `wrap_master_key` so a wrap error
145 // scrubs the fresh key instead of dropping a bare array.
146 let key = crypto::ZeroizeOnDrop(crypto::generate_master_key());
147 let envelope = crypto::wrap_master_key(&key.0, password)?;
148 Ok((key.0, envelope))
149 }
150 }
151 }
152
153 /// Run [`reencrypt_batch`](Self::reencrypt_batch) until the server reports no
154 /// entries left, bounded by [`MAX_ROTATION_ROUNDS`]. The batch set shrinks
155 /// every round under an honest server; the cap is the backstop against a
156 /// server that never reports it done, so the client fails with an error
157 /// instead of spinning forever.
158 async fn reencrypt_until_done(
159 &self,
160 rotation_id: Uuid,
161 old_key: &[u8; 32],
162 new_key: &[u8; 32],
163 ) -> Result<()> {
164 for _ in 0..MAX_ROTATION_ROUNDS {
165 if self.reencrypt_batch(rotation_id, old_key, new_key).await? {
166 return Ok(());
167 }
168 }
169 Err(crate::error::SyncKitError::Internal(
170 "key rotation did not converge: re-encrypt exceeded the round cap without the entry set draining".into(),
171 ))
172 }
173
174 /// Pull a batch of entries needing re-encryption, re-encrypt them, and push back.
175 /// Returns true if there are no more entries to process.
176 async fn reencrypt_batch(
177 &self,
178 rotation_id: Uuid,
179 old_key: &[u8; 32],
180 new_key: &[u8; 32],
181 ) -> Result<bool> {
182 // Pull entries needing re-encryption (starting from seq 0 each time,
183 // since the server filters by key_id != new_key_id)
184 let entries_resp = self.rotation_entries(rotation_id, 0).await?;
185
186 if entries_resp.entries.is_empty() {
187 return Ok(true);
188 }
189
190 let has_more = entries_resp.has_more;
191
192 // Re-encrypt each entry
193 let reencrypted: Vec<RotationBatchEntry> = entries_resp
194 .entries
195 .into_iter()
196 .map(|entry| {
197 let new_data = match entry.data {
198 Some(ref encrypted_value) => {
199 // Decrypt with old key, re-encrypt with new key, binding the
200 // same (table, row_id) AAD on both ends. A legacy untagged
201 // entry decrypts with empty AAD and re-emits as a v2 tagged,
202 // position-bound payload (opportunistic upgrade).
203 let ctx = crypto::AeadContext::entry(&entry.table, &entry.row_id);
204 let plaintext = crypto::decrypt_json_aad(encrypted_value, old_key, &ctx)?;
205 let reencrypted = crypto::encrypt_json_aad(&plaintext, new_key, &ctx)?;
206 Some(reencrypted)
207 }
208 None => None, // DELETE entries have no data
209 };
210 Ok(RotationBatchEntry {
211 seq: entry.seq,
212 data: new_data,
213 })
214 })
215 .collect::<Result<Vec<_>>>()?;
216
217 // Push re-encrypted batch
218 let count = reencrypted.len();
219 self.rotation_batch(rotation_id, reencrypted).await?;
220
221 tracing::debug!(count, "Re-encrypted batch submitted");
222
223 Ok(!has_more)
224 }
225
226 // ── HTTP helpers ──
227
228 async fn begin_rotation(
229 &self,
230 device_id: DeviceId,
231 new_encrypted_key: &str,
232 expected_key_version: i32,
233 ) -> Result<BeginRotationResponse> {
234 let token = self.require_token()?;
235 let url = format!(
236 "{}/api/v1/sync/keys/rotate",
237 self.config.server_url.trim_end_matches('/')
238 );
239
240 let body = Bytes::from(serde_json::to_vec(&BeginRotationRequest {
241 device_id,
242 new_encrypted_key: new_encrypted_key.to_string(),
243 expected_key_version,
244 })?);
245
246 self.retry_request_json(
247 Idempotency::IdempotentWrite {
248 on: "one rotation per (app_id, user_id), migration 089",
249 },
250 || {
251 let req = self
252 .http
253 .post(&url)
254 .bearer_auth(&token)
255 .header("content-type", "application/json")
256 .body(body.clone());
257 async move { check_response(req.send().await?).await }
258 },
259 )
260 .await
261 }
262
263 async fn rotation_entries(
264 &self,
265 rotation_id: Uuid,
266 after_seq: i64,
267 ) -> Result<RotationEntriesResponse> {
268 let token = self.require_token()?;
269 let url = format!(
270 "{}/api/v1/sync/keys/rotate/entries",
271 self.config.server_url.trim_end_matches('/')
272 );
273
274 let body = Bytes::from(serde_json::to_vec(&RotationEntriesRequest {
275 rotation_id,
276 after_seq,
277 })?);
278
279 self.retry_request_json(Idempotency::ReadOnly, || {
280 let req = self
281 .http
282 .post(&url)
283 .bearer_auth(&token)
284 .header("content-type", "application/json")
285 .body(body.clone());
286 async move { check_response(req.send().await?).await }
287 })
288 .await
289 }
290
291 async fn rotation_batch(
292 &self,
293 rotation_id: Uuid,
294 entries: Vec<RotationBatchEntry>,
295 ) -> Result<RotationBatchResponse> {
296 let token = self.require_token()?;
297 let url = format!(
298 "{}/api/v1/sync/keys/rotate/batch",
299 self.config.server_url.trim_end_matches('/')
300 );
301
302 let body = Bytes::from(serde_json::to_vec(&RotationBatchRequest {
303 rotation_id,
304 entries,
305 })?);
306
307 self.retry_request_json(
308 Idempotency::IdempotentWrite {
309 on: "seq (re-encrypt overwrites the same row)",
310 },
311 || {
312 let req = self
313 .http
314 .post(&url)
315 .bearer_auth(&token)
316 .header("content-type", "application/json")
317 .body(body.clone());
318 async move { check_response(req.send().await?).await }
319 },
320 )
321 .await
322 }
323
324 async fn complete_rotation(&self, rotation_id: Uuid) -> Result<()> {
325 let token = self.require_token()?;
326 let url = format!(
327 "{}/api/v1/sync/keys/rotate/complete",
328 self.config.server_url.trim_end_matches('/')
329 );
330
331 let body = Bytes::from(serde_json::to_vec(&CompleteRotationRequest {
332 rotation_id,
333 })?);
334
335 self.retry_request(Idempotency::IdempotentWrite { on: "rotation_id" }, || {
336 let req = self
337 .http
338 .post(&url)
339 .bearer_auth(&token)
340 .header("content-type", "application/json")
341 .body(body.clone());
342 async move { check_response(req.send().await?).await }
343 })
344 .await?;
345
346 Ok(())
347 }
348 }
349
350 #[cfg(test)]
351 mod tests {
352 use super::*;
353 use crate::types::{GetKeyResponse, PullChangeEntry};
354
355 #[test]
356 fn rotation_resume_reuses_committed_pending_key() {
357 // Simulate an interrupted rotation: the server holds a pending envelope
358 // wrapping key K. Resuming must recover K (not mint a fresh key), or
359 // entries already re-encrypted under K become undecryptable.
360 let committed_key = crypto::generate_master_key();
361 let pending_envelope = crypto::wrap_master_key(&committed_key, "pw").unwrap();
362 let pending = Some(PendingKeyInfo {
363 encrypted_key: pending_envelope.clone(),
364 key_id: 2,
365 });
366
367 let (resumed_key, envelope) = SyncKitClient::rotation_key_material(pending, "pw").unwrap();
368 assert_eq!(
369 resumed_key, committed_key,
370 "resume must adopt the committed key"
371 );
372 assert_eq!(
373 envelope, pending_envelope,
374 "resume must keep the committed envelope"
375 );
376 }
377
378 #[test]
379 fn rotation_fresh_start_mints_unwrappable_key() {
380 // No pending rotation: a fresh key is generated and its envelope unwraps
381 // back to that same key under the password.
382 let (fresh_key, envelope) = SyncKitClient::rotation_key_material(None, "pw").unwrap();
383 let recovered = crypto::unwrap_master_key(&envelope, "pw").unwrap();
384 assert_eq!(recovered, fresh_key);
385 }
386
387 #[test]
388 fn reencrypt_preserves_plaintext() {
389 // Simulate re-encryption: encrypt with old key, decrypt, re-encrypt with new key
390 let old_key = crypto::generate_master_key();
391 let new_key = crypto::generate_master_key();
392 let original = serde_json::json!({"title": "Test task", "priority": 3});
393
394 // Encrypt with old key (simulates existing sync_log entry)
395 let encrypted_old = crypto::encrypt_json(&original, &old_key).unwrap();
396
397 // Re-encrypt: decrypt with old, encrypt with new
398 let plaintext = crypto::decrypt_json(&encrypted_old, &old_key).unwrap();
399 let encrypted_new = crypto::encrypt_json(&plaintext, &new_key).unwrap();
400
401 // Verify: new key can decrypt to original data
402 let recovered = crypto::decrypt_json(&encrypted_new, &new_key).unwrap();
403 assert_eq!(recovered, original);
404
405 // Verify: old key cannot decrypt re-encrypted data
406 assert!(crypto::decrypt_json(&encrypted_new, &old_key).is_err());
407 }
408
409 #[test]
410 fn reencrypt_null_data_stays_null() {
411 // DELETE entries have no data, rotation should preserve this
412 let old_key = crypto::generate_master_key();
413 let new_key = crypto::generate_master_key();
414
415 // No data to re-encrypt
416 let data: Option<serde_json::Value> = None;
417 let reencrypted = match data {
418 Some(ref val) => {
419 let plaintext = crypto::decrypt_json(val, &old_key).unwrap();
420 Some(crypto::encrypt_json(&plaintext, &new_key).unwrap())
421 }
422 None => None,
423 };
424
425 assert!(reencrypted.is_none());
426 }
427
428 #[test]
429 fn rotation_request_types_serialize() {
430 let req = BeginRotationRequest {
431 device_id: DeviceId::new(Uuid::new_v4()),
432 new_encrypted_key: "envelope-json".to_string(),
433 expected_key_version: 1,
434 };
435 let json = serde_json::to_string(&req).unwrap();
436 assert!(json.contains("expected_key_version"));
437
438 let req = RotationBatchRequest {
439 rotation_id: Uuid::new_v4(),
440 entries: vec![
441 RotationBatchEntry {
442 seq: 1,
443 data: Some(serde_json::json!("encrypted")),
444 },
445 RotationBatchEntry { seq: 2, data: None },
446 ],
447 };
448 let json = serde_json::to_string(&req).unwrap();
449 assert!(json.contains("rotation_id"));
450 assert!(json.contains("\"seq\":1"));
451 }
452
453 #[test]
454 fn rotation_response_types_deserialize() {
455 let json = r#"{"rotation_id": "550e8400-e29b-41d4-a716-446655440000", "target_seq": 100, "new_key_id": 2}"#;
456 let resp: BeginRotationResponse = serde_json::from_str(json).unwrap();
457 assert_eq!(resp.target_seq, 100);
458 assert_eq!(resp.new_key_id, 2);
459
460 let json = r#"{"entries": [{"seq": 1, "table": "tasks", "row_id": "r1", "data": "encrypted"}, {"seq": 2, "table": "tasks", "row_id": "r2", "data": null}], "has_more": true}"#;
461 let resp: RotationEntriesResponse = serde_json::from_str(json).unwrap();
462 assert_eq!(resp.entries.len(), 2);
463 assert!(resp.has_more);
464 assert!(resp.entries[0].data.is_some());
465 assert_eq!(resp.entries[0].table, "tasks");
466 assert_eq!(resp.entries[0].row_id, "r1");
467 assert!(resp.entries[1].data.is_none());
468 }
469
470 #[test]
471 fn pending_key_info_deserializes() {
472 let json = r#"{
473 "encrypted_key": "envelope",
474 "key_version": 2,
475 "key_id": 1,
476 "pending_key": {"encrypted_key": "new-envelope", "key_id": 2}
477 }"#;
478 let resp: GetKeyResponse = serde_json::from_str(json).unwrap();
479 assert!(resp.pending_key.is_some());
480 let pending = resp.pending_key.unwrap();
481 assert_eq!(pending.key_id, 2);
482 assert_eq!(pending.encrypted_key, "new-envelope");
483 }
484
485 #[test]
486 fn get_key_response_without_pending_key() {
487 let json = r#"{"encrypted_key": "envelope", "key_version": 1}"#;
488 let resp: GetKeyResponse = serde_json::from_str(json).unwrap();
489 assert!(resp.pending_key.is_none());
490 assert_eq!(resp.key_id, None); // backward compat
491 }
492
493 #[test]
494 fn pull_change_entry_with_key_id() {
495 let device_id = Uuid::new_v4();
496 let json = format!(
497 r#"{{"seq": 1, "device_id": "{device_id}", "table": "t", "op": "INSERT", "row_id": "r", "timestamp": "2025-06-01T12:00:00Z", "data": null, "key_id": 2}}"#
498 );
499 let entry: PullChangeEntry = serde_json::from_str(&json).unwrap();
500 assert_eq!(entry.key_id, Some(2));
501 }
502
503 #[test]
504 fn pull_change_entry_without_key_id_defaults_none() {
505 let device_id = Uuid::new_v4();
506 let json = format!(
507 r#"{{"seq": 1, "device_id": "{device_id}", "table": "t", "op": "INSERT", "row_id": "r", "timestamp": "2025-06-01T12:00:00Z", "data": null}}"#
508 );
509 let entry: PullChangeEntry = serde_json::from_str(&json).unwrap();
510 assert_eq!(entry.key_id, None);
511 }
512 }
513