Skip to main content

max / synckit

Split the integration suite into domain modules, test the import direction tests/integration.rs was one 4,201-line file holding 120 tests, so nobody could find the existing test for a behaviour and would write a second one beside it. It is now tests/integration/ with common.rs plus 11 domain modules, largest 693 lines. The [[test]] target is explicit because a target rooted at tests/integration.rs resolves `mod common;` to tests/common.rs rather than into the directory beside it. Adds five tests to store/config.rs for the import direction of the posture filter. Every existing test drove the export path, a local write reaching the changelog. Nothing exercised inbound, which is the direction fuzz-2026-07-21 #3 was about: a hostile server steering a local write root across the boundary. The guard was already correct, just unverified. Same tests before and after the split: 118 with --all-features, 120 with keychain off (the two gated rotation tests).
Co-Authored-By
Claude Opus 5 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-08-04 21:09 UTC
Signed with PGP, not checked
Commit: f007473983627bafe830bd6abd0b3470b8bc3015
Parent: 5e88f2d
17 files changed, +4345 insertions, -500 deletions
M .gitignore +4
@@ -1,3 +1,7 @@
1 1 /target
2 2 **/target
3 3 Cargo.lock
4 +
5 + # cargo-mutants run artifacts
6 + mutants.out/
7 + mutants.out.old/
@@ -98,6 +98,13 @@
98 98 [target.'cfg(target_os = "ios")'.dependencies]
99 99 apple-native-keyring-store = { version = "1.0.1", features = ["protected"], optional = true }
100 100
101 + [[test]]
102 + # Explicit path: a test target rooted at `tests/integration.rs` would resolve
103 + # `mod common;` to `tests/common.rs`, not into the directory. Rooting it at
104 + # `tests/integration/main.rs` puts the modules where they belong, beside it.
105 + name = "integration"
106 + path = "tests/integration/main.rs"
107 +
101 108 [dev-dependencies]
102 109 wiremock = "0.6"
103 110 # reqwest is built `rustls-no-provider`, so real consumers install a rustls crypto
@@ -95,7 +95,9 @@
95 95 #[cfg(test)]
96 96 mod tests {
97 97 use super::*;
98 + use crate::store::apply::apply_remote_changes;
98 99 use crate::store::schema::SyncSchema;
100 + use crate::types::{ChangeEntry, ChangeOp, hlc_legacy_floor};
99 101 use synckit_config::{ConfigStore, Posture};
100 102
101 103 const SPEC: ConfigSpec = ConfigSpec::new(
@@ -190,6 +192,156 @@
190 192 );
191 193 }
192 194
195 + // ── the import direction ──
196 + //
197 + // Everything above drives the export trigger: a local write, and whether it
198 + // reaches the changelog. The predicate is meant to hold symmetrically on
199 + // import (`apply::is_excluded` evaluates the same text against the incoming
200 + // row), and that is the direction the fuzz finding in the module header was
201 + // about: a hostile server pushing `mirror_path` to steer a local write root.
202 + // These pin that half.
203 +
204 + /// An inbound change for `table`, as the server would send it.
205 + fn inbound(key: &str, value: &str) -> ChangeEntry {
206 + ChangeEntry {
207 + table: "app_config".into(),
208 + op: ChangeOp::Insert,
209 + row_id: key.into(),
210 + timestamp: chrono::Utc::now(),
211 + hlc: hlc_legacy_floor(),
212 + data: Some(serde_json::json!({ "key": key, "value": value })),
213 + extra: serde_json::Map::default(),
214 + }
215 + }
216 +
217 + fn stored(conn: &Connection, store: &ConfigStore, key: &str) -> Option<String> {
218 + store.get(conn, key).unwrap()
219 + }
220 +
221 + // Positive control: without this, the two rejection tests below would pass
222 + // just as well if apply were broken outright.
223 + #[test]
224 + fn an_inbound_synced_key_is_applied() {
225 + let (mut conn, store) = wired();
226 + let outcome = apply_remote_changes(
227 + &mut conn,
228 + &SyncSchema::new(vec![config_sync_table(&SPEC)]),
229 + &[inbound("theme", "akari-night")],
230 + "",
231 + )
232 + .unwrap();
233 +
234 + assert_eq!(outcome.applied, 1);
235 + assert_eq!(outcome.filtered, 0);
236 + assert_eq!(
237 + stored(&conn, &store, "theme").as_deref(),
238 + Some("akari-night")
239 + );
240 + }
241 +
242 + // The server does not get to write a Local key. `mirror_path` is a filesystem
243 + // root; accepting one from the wire is the write-root steer (fuzz-2026-07-21 #3).
244 + #[test]
245 + fn an_inbound_local_key_is_filtered_not_applied() {
246 + let (mut conn, store) = wired();
247 + store
248 + .set(&conn, "mirror_path", "/home/max/samples")
249 + .unwrap();
250 +
251 + let outcome = apply_remote_changes(
252 + &mut conn,
253 + &SyncSchema::new(vec![config_sync_table(&SPEC)]),
254 + &[inbound("mirror_path", "/tmp/attacker")],
255 + "",
256 + )
257 + .unwrap();
258 +
259 + assert_eq!(outcome.applied, 0);
260 + assert_eq!(
261 + outcome.filtered, 1,
262 + "policy, not failure: the row is dropped, never retried",
263 + );
264 + assert!(outcome.rejected.is_empty() && outcome.deferred.is_empty());
265 + assert_eq!(
266 + stored(&conn, &store, "mirror_path").as_deref(),
267 + Some("/home/max/samples"),
268 + "the local value survives an inbound attempt to overwrite it",
269 + );
270 + }
271 +
272 + // The allowlist on import: a key the spec never classified has no policy row,
273 + // so the predicate cannot admit it and the server cannot introduce it.
274 + #[test]
275 + fn an_inbound_undeclared_key_is_filtered_not_applied() {
276 + let (mut conn, store) = wired();
277 + let outcome = apply_remote_changes(
278 + &mut conn,
279 + &SyncSchema::new(vec![config_sync_table(&SPEC)]),
280 + &[inbound("some_new_local_path", "/etc/secret")],
281 + "",
282 + )
283 + .unwrap();
284 +
285 + assert_eq!(outcome.applied, 0);
286 + assert_eq!(outcome.filtered, 1);
287 + assert_eq!(
288 + stored(&conn, &store, "some_new_local_path"),
289 + None,
290 + "an unclassified key cannot be introduced from the wire",
291 + );
292 + }
293 +
294 + // A delete is the other half of the write: dropping a Local key is as much a
295 + // local mutation as setting one, so the guard must catch it too.
296 + #[test]
297 + fn an_inbound_delete_of_a_local_key_is_filtered() {
298 + let (mut conn, store) = wired();
299 + store
300 + .set(&conn, "mirror_path", "/home/max/samples")
301 + .unwrap();
302 +
303 + let outcome = apply_remote_changes(
304 + &mut conn,
305 + &SyncSchema::new(vec![config_sync_table(&SPEC)]),
306 + &[ChangeEntry {
307 + op: ChangeOp::Delete,
308 + ..inbound("mirror_path", "/home/max/samples")
309 + }],
310 + "",
311 + )
312 + .unwrap();
313 +
314 + assert_eq!(outcome.applied, 0);
315 + assert_eq!(outcome.filtered, 1);
316 + assert_eq!(
317 + stored(&conn, &store, "mirror_path").as_deref(),
318 + Some("/home/max/samples"),
319 + "the server cannot delete a key it was never allowed to write",
320 + );
321 + }
322 +
323 + // Re-seeding a narrower spec closes the import door as well as the export one:
324 + // a key dropped from the spec stops being writable from the wire.
325 + #[test]
326 + fn reseeding_a_narrower_spec_stops_the_dropped_key_importing() {
327 + let (mut conn, store) = wired();
328 + const NARROWER: ConfigSpec =
329 + ConfigSpec::new("app_config", &[("sidebar_visible", Posture::Synced)]);
330 + install_policy(&mut conn, &NARROWER).unwrap();
331 +
332 + let outcome = apply_remote_changes(
333 + &mut conn,
334 + &SyncSchema::new(vec![config_sync_table(&SPEC)]),
335 + &[inbound("theme", "akari-night")],
336 + "",
337 + )
338 + .unwrap();
339 +
340 + assert_eq!(outcome.applied, 0);
341 + assert_eq!(outcome.filtered, 1);
342 + assert_eq!(stored(&conn, &store, "theme"), None);
343 + }
344 +
193 345 #[test]
194 346 fn install_policy_is_idempotent() {
195 347 let (mut conn, _store) = wired();
@@ -1,4201 +1,0 @@
1 - //! Integration tests using wiremock to simulate the MNW SyncKit server.
2 - //!
3 - //! These tests verify the full HTTP round-trip including retry behavior,
4 - //! encryption/decryption, and error classification.
5 -
6 - use std::sync::Arc;
7 -
8 - use base64::Engine;
9 - use chrono::Utc;
10 - use serde_json::json;
11 - use sha2::Digest;
12 - use uuid::Uuid;
13 - use wiremock::matchers::{method, path};
14 - use wiremock::{Mock, MockServer, ResponseTemplate};
15 -
16 - use std::time::Duration;
17 - use synckit_client::{
18 - AppId, ChangeEntry, ChangeOp, DeviceId, Hlc, SyncKitClient, SyncKitConfig, SyncKitError, UserId,
19 - };
20 -
21 - // ── Helpers ──
22 -
23 - fn fake_jwt(exp: i64) -> String {
24 - let header =
25 - base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(r#"{"alg":"HS256","typ":"JWT"}"#);
26 - let payload = json!({
27 - "sub": "550e8400-e29b-41d4-a716-446655440000",
28 - "app": "6ba7b810-9dad-11d1-80b4-00c04fd430c8",
29 - "exp": exp,
30 - "iat": exp - 3600,
31 - });
32 - let payload_b64 =
33 - base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(payload.to_string().as_bytes());
34 - let sig = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(b"fake-signature");
35 - format!("{header}.{payload_b64}.{sig}")
36 - }
37 -
38 - fn fresh_token() -> String {
39 - fake_jwt(Utc::now().timestamp() + 3600)
40 - }
41 -
42 - fn test_ids() -> (UserId, AppId) {
43 - (
44 - UserId::new(Uuid::parse_str("550e8400-e29b-41d4-a716-446655440000").unwrap()),
45 - AppId::new(Uuid::parse_str("6ba7b810-9dad-11d1-80b4-00c04fd430c8").unwrap()),
46 - )
47 - }
48 -
49 - /// Install the rustls crypto provider once. reqwest is built `rustls-no-provider`,
50 - /// so a real consumer app installs one at startup (audiofiles installs ring); these
51 - /// tests have no such app, so they install ring themselves before building a client.
52 - fn ensure_crypto_provider() {
53 - static PROVIDER: std::sync::Once = std::sync::Once::new();
54 - PROVIDER.call_once(|| {
55 - // Err means a provider is already installed, which is the outcome we want.
56 - let _ = rustls::crypto::ring::default_provider().install_default();
57 - });
58 - }
59 -
60 - fn client_for(server: &MockServer) -> SyncKitClient {
61 - ensure_crypto_provider();
62 - SyncKitClient::new(SyncKitConfig {
63 - server_url: server.uri(),
64 - api_key: "test-api-key".to_string(),
65 - })
66 - }
67 -
68 - fn authed_client(server: &MockServer) -> SyncKitClient {
69 - let client = client_for(server);
70 - let (user_id, app_id) = test_ids();
71 - client.restore_session(&fresh_token(), user_id, app_id);
72 - client
73 - }
74 -
75 - fn auth_response_json() -> serde_json::Value {
76 - let (user_id, app_id) = test_ids();
77 - json!({
78 - "token": fresh_token(),
79 - "user_id": user_id,
80 - "app_id": app_id,
81 - })
82 - }
83 -
84 - fn device_json() -> serde_json::Value {
85 - let (user_id, app_id) = test_ids();
86 - json!({
87 - "id": Uuid::new_v4(),
88 - "app_id": app_id,
89 - "user_id": user_id,
90 - "device_name": "Test Device",
91 - "platform": "test",
92 - "last_seen_at": "2025-01-01T00:00:00Z",
93 - "created_at": "2025-01-01T00:00:00Z",
94 - })
95 - }
96 -
97 - // ── Response body cap (DoS) ──
98 -
99 - #[tokio::test]
100 - async fn oversized_response_body_is_capped_not_buffered() {
101 - // A hostile/buggy server streams a control-plane body far larger than the
102 - // 8 MiB cap. The client must reject it (the cap fast-rejects on the honest
103 - // Content-Length) instead of buffering it into memory and OOMing.
104 - let server = MockServer::start().await;
105 - let huge = vec![b'x'; 9 * 1024 * 1024];
106 - Mock::given(method("GET"))
107 - .and(path("/api/v1/sync/status"))
108 - .respond_with(ResponseTemplate::new(200).set_body_bytes(huge))
109 - .mount(&server)
110 - .await;
111 -
112 - let client = authed_client(&server);
113 - let err = client.status().await.unwrap_err();
114 - assert!(
115 - matches!(err, SyncKitError::Internal(ref m) if m.contains("cap")),
116 - "expected the body cap to reject the oversized response, got: {err:?}"
117 - );
118 - }
119 -
120 - // ── Auth flow ──
121 -
122 - #[tokio::test]
123 - async fn authenticate_success_stores_session() {
124 - let server = MockServer::start().await;
125 -
126 - Mock::given(method("POST"))
127 - .and(path("/api/v1/sync/auth"))
128 - .respond_with(ResponseTemplate::new(200).set_body_json(auth_response_json()))
129 - .mount(&server)
130 - .await;
131 -
132 - let client = client_for(&server);
133 - let (user_id, app_id) = client
134 - .authenticate("user@test.com", "password", "test-key")
135 - .await
136 - .unwrap();
137 -
138 - let info = client.session_info().expect("session stored");
139 - assert_eq!(info.user_id, user_id);
140 - assert_eq!(info.app_id, app_id);
141 - }
142 -
143 - #[tokio::test]
144 - async fn authenticate_wrong_password_no_retry() {
145 - let server = MockServer::start().await;
146 -
147 - Mock::given(method("POST"))
148 - .and(path("/api/v1/sync/auth"))
149 - .respond_with(ResponseTemplate::new(401).set_body_string("Unauthorized"))
150 - .expect(1) // Must be called exactly once (no retry)
151 - .mount(&server)
152 - .await;
153 -
154 - let client = client_for(&server);
155 - let err = client
156 - .authenticate("user@test.com", "wrong", "test-key")
157 - .await
158 - .unwrap_err();
159 -
160 - assert!(
161 - matches!(err, SyncKitError::Server { status: 401, .. }),
162 - "Expected 401 error, got: {err:?}"
163 - );
164 - }
165 -
166 - #[tokio::test]
167 - async fn authenticate_retries_on_503() {
168 - let server = MockServer::start().await;
169 -
170 - Mock::given(method("POST"))
171 - .and(path("/api/v1/sync/auth"))
172 - .respond_with(ResponseTemplate::new(503).set_body_string("Service Unavailable"))
173 - .up_to_n_times(1)
174 - .mount(&server)
175 - .await;
176 -
177 - Mock::given(method("POST"))
178 - .and(path("/api/v1/sync/auth"))
179 - .respond_with(ResponseTemplate::new(200).set_body_json(auth_response_json()))
180 - .mount(&server)
181 - .await;
182 -
183 - let client = client_for(&server);
184 - let result = client
185 - .authenticate("user@test.com", "password", "test-key")
186 - .await;
187 - assert!(result.is_ok(), "Should succeed after retry: {result:?}");
188 - }
189 -
190 - #[tokio::test]
191 - async fn authenticate_with_code_success() {
192 - let server = MockServer::start().await;
193 -
194 - let (user_id, app_id) = test_ids();
195 - Mock::given(method("POST"))
196 - .and(path("/oauth/token"))
197 - .respond_with(ResponseTemplate::new(200).set_body_json(json!({
198 - "access_token": fresh_token(),
199 - "token_type": "Bearer",
200 - "expires_in": 3600,
201 - "user_id": user_id,
202 - "app_id": app_id,
203 - })))
204 - .mount(&server)
205 - .await;
206 -
207 - let client = client_for(&server);
208 - let (uid, aid) = client
209 - .authenticate_with_code("auth-code", "verifier", 8080, "test-key")
210 - .await
211 - .unwrap();
212 -
213 - assert_eq!(uid, user_id);
214 - assert_eq!(aid, app_id);
215 - assert!(client.session_info().is_some());
216 - }
217 -
218 - // ── Device management ──
219 -
220 - #[tokio::test]
221 - async fn register_device_success() {
222 - let server = MockServer::start().await;
223 -
224 - Mock::given(method("POST"))
225 - .and(path("/api/v1/sync/devices"))
226 - .respond_with(ResponseTemplate::new(200).set_body_json(device_json()))
227 - .mount(&server)
228 - .await;
229 -
230 - let client = authed_client(&server);
231 - let device = client.register_device("MacBook", "macos").await.unwrap();
232 - assert_eq!(device.device_name, "Test Device");
233 - }
234 -
235 - #[tokio::test]
236 - async fn register_device_retries_on_transient() {
237 - let server = MockServer::start().await;
238 -
239 - Mock::given(method("POST"))
240 - .and(path("/api/v1/sync/devices"))
241 - .respond_with(ResponseTemplate::new(502).set_body_string("Bad Gateway"))
242 - .up_to_n_times(1)
243 - .mount(&server)
244 - .await;
245 -
246 - Mock::given(method("POST"))
247 - .and(path("/api/v1/sync/devices"))
248 - .respond_with(ResponseTemplate::new(200).set_body_json(device_json()))
249 - .mount(&server)
250 - .await;
251 -
252 - let client = authed_client(&server);
253 - let result = client.register_device("MacBook", "macos").await;
254 - assert!(result.is_ok(), "Should succeed after retry: {result:?}");
255 - }
256 -
257 - #[tokio::test]
258 - async fn list_devices_success() {
259 - let server = MockServer::start().await;
260 -
261 - Mock::given(method("GET"))
262 - .and(path("/api/v1/sync/devices"))
263 - .respond_with(ResponseTemplate::new(200).set_body_json(json!([device_json()])))
264 - .mount(&server)
265 - .await;
266 -
267 - let client = authed_client(&server);
268 - let devices = client.list_devices().await.unwrap();
269 - assert_eq!(devices.len(), 1);
270 - assert_eq!(devices[0].device_name, "Test Device");
271 - }
272 -
273 - // ── Push / Pull with encryption ──
274 -
275 - #[tokio::test]
276 - async fn push_encrypts_data() {
277 - let server = MockServer::start().await;
278 -
279 - Mock::given(method("POST"))
280 - .and(path("/api/v1/sync/push"))
281 - .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 1})))
282 - .mount(&server)
283 - .await;
284 -
285 - let client = authed_client(&server);
286 - let key = synckit_client::crypto::generate_master_key();
287 - client.set_master_key_raw(key);
288 -
289 - let device_id = DeviceId::new(Uuid::new_v4());
290 - let cursor = client
291 - .push(
292 - device_id,
293 - vec![ChangeEntry {
294 - table: "tasks".into(),
295 - op: ChangeOp::Insert,
296 - row_id: "row-1".into(),
297 - timestamp: Utc::now(),
298 - hlc: Hlc::zero(DeviceId::nil()),
299 - data: Some(json!({"title": "Secret task"})),
300 - extra: serde_json::Map::default(),
301 - }],
302 - )
303 - .await
304 - .unwrap();
305 -
306 - assert_eq!(cursor, 1);
307 -
308 - // Verify the request body was sent with encrypted data (not plaintext)
309 - let requests = server.received_requests().await.unwrap();
310 - let push_req = requests
311 - .iter()
312 - .find(|r| r.url.path() == "/api/v1/sync/push")
313 - .unwrap();
314 - let body: serde_json::Value = serde_json::from_slice(&push_req.body).unwrap();
315 - let wire_data = body["changes"][0]["data"].as_str().unwrap();
316 - assert!(
317 - !wire_data.contains("Secret task"),
318 - "Plaintext should not appear on the wire"
319 - );
320 - }
321 -
322 - #[tokio::test]
323 - async fn pull_decrypts_data() {
324 - let server = MockServer::start().await;
325 -
326 - let client = authed_client(&server);
327 - let key = synckit_client::crypto::generate_master_key();
328 - client.set_master_key_raw(key);
329 -
330 - // Encrypt a value to simulate what the server would return
331 - let plaintext = json!({"title": "Decrypted task"});
332 - let encrypted = synckit_client::crypto::encrypt_json(&plaintext, &key).unwrap();
333 -
334 - let device_id = DeviceId::new(Uuid::new_v4());
335 - Mock::given(method("POST"))
336 - .and(path("/api/v1/sync/pull"))
337 - .respond_with(ResponseTemplate::new(200).set_body_json(json!({
338 - "changes": [{
339 - "seq": 1,
340 - "device_id": device_id,
341 - "table": "tasks",
342 - "op": "INSERT",
343 - "row_id": "row-1",
344 - "timestamp": "2025-06-01T12:00:00Z",
345 - "data": encrypted,
346 - }],
347 - "cursor": 1,
348 - "has_more": false,
349 - })))
350 - .mount(&server)
351 - .await;
352 -
353 - let (changes, cursor, has_more) = client.pull(device_id, 0).await.unwrap();
354 - assert_eq!(changes.len(), 1);
355 - assert_eq!(cursor, 1);
356 - assert!(!has_more);
357 - assert_eq!(changes[0].data.as_ref().unwrap(), &plaintext);
358 - }
359 -
360 - #[tokio::test]
361 - async fn push_retries_on_503() {
362 - let server = MockServer::start().await;
363 -
364 - Mock::given(method("POST"))
365 - .and(path("/api/v1/sync/push"))
366 - .respond_with(ResponseTemplate::new(503))
367 - .up_to_n_times(1)
368 - .mount(&server)
369 - .await;
370 -
371 - Mock::given(method("POST"))
372 - .and(path("/api/v1/sync/push"))
373 - .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 5})))
374 - .mount(&server)
375 - .await;
376 -
377 - let client = authed_client(&server);
378 - let key = synckit_client::crypto::generate_master_key();
379 - client.set_master_key_raw(key);
380 -
381 - let cursor = client
382 - .push(DeviceId::new(Uuid::new_v4()), vec![])
383 - .await
384 - .unwrap();
385 - assert_eq!(cursor, 5);
386 - }
387 -
388 - #[tokio::test]
389 - async fn push_fails_immediately_on_401() {
390 - let server = MockServer::start().await;
391 -
392 - Mock::given(method("POST"))
393 - .and(path("/api/v1/sync/push"))
394 - .respond_with(ResponseTemplate::new(401).set_body_string("Unauthorized"))
395 - .expect(1)
396 - .mount(&server)
397 - .await;
398 -
399 - let client = authed_client(&server);
400 - let key = synckit_client::crypto::generate_master_key();
401 - client.set_master_key_raw(key);
402 -
403 - let err = client
404 - .push(DeviceId::new(Uuid::new_v4()), vec![])
405 - .await
406 - .unwrap_err();
407 - assert!(matches!(err, SyncKitError::Server { status: 401, .. }));
408 - }
409 -
410 - #[tokio::test]
411 - async fn pull_with_has_more_pagination() {
412 - let server = MockServer::start().await;
413 -
414 - let client = authed_client(&server);
415 - let key = synckit_client::crypto::generate_master_key();
416 - client.set_master_key_raw(key);
417 -
418 - let device_id = DeviceId::new(Uuid::new_v4());
419 -
420 - // First pull: has_more = true
421 - Mock::given(method("POST"))
422 - .and(path("/api/v1/sync/pull"))
423 - .respond_with(ResponseTemplate::new(200).set_body_json(json!({
424 - "changes": [],
425 - "cursor": 50,
426 - "has_more": true,
427 - })))
428 - .up_to_n_times(1)
429 - .mount(&server)
430 - .await;
431 -
432 - let (changes, cursor, has_more) = client.pull(device_id, 0).await.unwrap();
433 - assert!(changes.is_empty());
434 - assert_eq!(cursor, 50);
435 - assert!(has_more);
436 -
437 - // Second pull from cursor 50: has_more = false
438 - Mock::given(method("POST"))
439 - .and(path("/api/v1/sync/pull"))
440 - .respond_with(ResponseTemplate::new(200).set_body_json(json!({
441 - "changes": [],
442 - "cursor": 100,
443 - "has_more": false,
444 - })))
445 - .mount(&server)
446 - .await;
447 -
448 - let (_, cursor2, has_more2) = client.pull(device_id, 50).await.unwrap();
449 - assert_eq!(cursor2, 100);
450 - assert!(!has_more2);
451 - }
452 -
453 - // ── Blob operations ──
454 -
455 - #[tokio::test]
456 - async fn blob_upload_url_success() {
457 - let server = MockServer::start().await;
458 -
459 - Mock::given(method("POST"))
460 - .and(path("/api/v1/sync/blobs/upload"))
461 - .respond_with(ResponseTemplate::new(200).set_body_json(json!({
462 - "upload_url": "https://s3.example.com/put",
463 - "already_exists": false,
464 - })))
465 - .mount(&server)
466 - .await;
467 -
468 - let client = authed_client(&server);
469 - let resp = client.blob_upload_url("sha256-abc", 1024).await.unwrap();
470 - assert_eq!(resp.upload_url, "https://s3.example.com/put");
471 - assert!(!resp.already_exists);
472 - }
473 -
474 - #[tokio::test]
475 - async fn blob_upload_url_declares_the_length_the_put_will_carry() {
476 - // The server signs the declared size into the presigned URL as
477 - // Content-Length, a SignedHeader, so declaring anything other than the
478 - // exact ciphertext length makes the PUT fail SigV4. The caller passes the
479 - // plaintext size it sees on disk; the SDK converts. This test pins the two
480 - // halves together, which is the only place the mismatch would show up:
481 - // wiremock does not verify signatures, and the server's own tests use an
482 - // in-memory backend that does not sign at all.
483 - let server = MockServer::start().await;
484 -
485 - let upload_path = "/s3/sized-upload";
486 - Mock::given(method("POST"))
487 - .and(path("/api/v1/sync/blobs/upload"))
488 - .respond_with(ResponseTemplate::new(200).set_body_json(json!({
489 - "upload_url": format!("{}{upload_path}", server.uri()),
490 - "already_exists": false,
491 - })))
492 - .mount(&server)
493 - .await;
494 - Mock::given(method("PUT"))
495 - .and(path(upload_path))
496 - .respond_with(ResponseTemplate::new(200))
497 - .mount(&server)
498 - .await;
499 -
500 - let client = authed_client(&server);
Lines truncated
@@ -1,0 +1,352 @@
1 + //! Authentication, token handling, and session lifecycle.
2 + //!
3 + //! The credential exchange (`authenticate`, `authenticate_with_code`), what the
4 + //! client does with a JWT that is expired or near expiry, and the session
5 + //! transitions: restore, clear, re-authenticate over a live session.
6 +
7 + use crate::common::*;
8 +
9 + // ── Auth flow ──
10 +
11 + #[tokio::test]
12 + async fn authenticate_success_stores_session() {
13 + let server = MockServer::start().await;
14 +
15 + Mock::given(method("POST"))
16 + .and(path("/api/v1/sync/auth"))
17 + .respond_with(ResponseTemplate::new(200).set_body_json(auth_response_json()))
18 + .mount(&server)
19 + .await;
20 +
21 + let client = client_for(&server);
22 + let (user_id, app_id) = client
23 + .authenticate("user@test.com", "password", "test-key")
24 + .await
25 + .unwrap();
26 +
27 + let info = client.session_info().expect("session stored");
28 + assert_eq!(info.user_id, user_id);
29 + assert_eq!(info.app_id, app_id);
30 + }
31 +
32 + #[tokio::test]
33 + async fn authenticate_wrong_password_no_retry() {
34 + let server = MockServer::start().await;
35 +
36 + Mock::given(method("POST"))
37 + .and(path("/api/v1/sync/auth"))
38 + .respond_with(ResponseTemplate::new(401).set_body_string("Unauthorized"))
39 + .expect(1) // Must be called exactly once (no retry)
40 + .mount(&server)
41 + .await;
42 +
43 + let client = client_for(&server);
44 + let err = client
45 + .authenticate("user@test.com", "wrong", "test-key")
46 + .await
47 + .unwrap_err();
48 +
49 + assert!(
50 + matches!(err, SyncKitError::Server { status: 401, .. }),
51 + "Expected 401 error, got: {err:?}"
52 + );
53 + }
54 +
55 + #[tokio::test]
56 + async fn authenticate_retries_on_503() {
57 + let server = MockServer::start().await;
58 +
59 + Mock::given(method("POST"))
60 + .and(path("/api/v1/sync/auth"))
61 + .respond_with(ResponseTemplate::new(503).set_body_string("Service Unavailable"))
62 + .up_to_n_times(1)
63 + .mount(&server)
64 + .await;
65 +
66 + Mock::given(method("POST"))
67 + .and(path("/api/v1/sync/auth"))
68 + .respond_with(ResponseTemplate::new(200).set_body_json(auth_response_json()))
69 + .mount(&server)
70 + .await;
71 +
72 + let client = client_for(&server);
73 + let result = client
74 + .authenticate("user@test.com", "password", "test-key")
75 + .await;
76 + assert!(result.is_ok(), "Should succeed after retry: {result:?}");
77 + }
78 +
79 + #[tokio::test]
80 + async fn authenticate_with_code_success() {
81 + let server = MockServer::start().await;
82 +
83 + let (user_id, app_id) = test_ids();
84 + Mock::given(method("POST"))
85 + .and(path("/oauth/token"))
86 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
87 + "access_token": fresh_token(),
88 + "token_type": "Bearer",
89 + "expires_in": 3600,
90 + "user_id": user_id,
91 + "app_id": app_id,
92 + })))
93 + .mount(&server)
94 + .await;
95 +
96 + let client = client_for(&server);
97 + let (uid, aid) = client
98 + .authenticate_with_code("auth-code", "verifier", 8080, "test-key")
99 + .await
100 + .unwrap();
101 +
102 + assert_eq!(uid, user_id);
103 + assert_eq!(aid, app_id);
104 + assert!(client.session_info().is_some());
105 + }
106 +
107 + // ── Token handling ──
108 +
109 + #[tokio::test]
110 + async fn expired_jwt_returns_token_expired() {
111 + let server = MockServer::start().await;
112 + let client = client_for(&server);
113 + let (user_id, app_id) = test_ids();
114 +
115 + let expired = fake_jwt(Utc::now().timestamp() - 3600);
116 + client.restore_session(&expired, user_id, app_id);
117 +
118 + let err = client.status().await.unwrap_err();
119 + assert!(
120 + matches!(err, SyncKitError::TokenExpired),
121 + "Expected TokenExpired, got: {err:?}"
122 + );
123 + }
124 +
125 + #[tokio::test]
126 + async fn near_expiry_jwt_returns_token_expired() {
127 + let server = MockServer::start().await;
128 + let client = client_for(&server);
129 + let (user_id, app_id) = test_ids();
130 +
131 + // Token expires in 10 seconds (within 30-second buffer)
132 + let near_expiry = fake_jwt(Utc::now().timestamp() + 10);
133 + client.restore_session(&near_expiry, user_id, app_id);
134 +
135 + let err = client.status().await.unwrap_err();
136 + assert!(
137 + matches!(err, SyncKitError::TokenExpired),
138 + "Expected TokenExpired, got: {err:?}"
139 + );
140 + }
141 +
142 + // ── Session management ──
143 +
144 + #[tokio::test]
145 + async fn restore_then_clear_session() {
146 + let server = MockServer::start().await;
147 +
148 + Mock::given(method("GET"))
149 + .and(path("/api/v1/sync/status"))
150 + .respond_with(
151 + ResponseTemplate::new(200)
152 + .set_body_json(json!({"total_changes": 0, "latest_cursor": null})),
153 + )
154 + .mount(&server)
155 + .await;
156 +
157 + let client = authed_client(&server);
158 +
159 + // Should work while authenticated
160 + let status = client.status().await.unwrap();
161 + assert_eq!(status.total_changes, 0);
162 +
163 + // Clear session
164 + client.clear_session();
165 +
166 + // Now should fail
167 + let err = client.status().await.unwrap_err();
168 + assert!(matches!(err, SyncKitError::NotAuthenticated));
169 + }
170 +
171 + #[tokio::test]
172 + async fn status_without_auth_returns_not_authenticated() {
173 + let server = MockServer::start().await;
174 + let client = client_for(&server);
175 +
176 + let err = client.status().await.unwrap_err();
177 + assert!(matches!(err, SyncKitError::NotAuthenticated));
178 + }
179 +
180 + #[tokio::test]
181 + async fn push_without_auth_returns_not_authenticated() {
182 + let server = MockServer::start().await;
183 + let client = client_for(&server);
184 +
185 + let err = client
186 + .push(DeviceId::new(Uuid::new_v4()), vec![])
187 + .await
188 + .unwrap_err();
189 + assert!(matches!(err, SyncKitError::NotAuthenticated));
190 + }
191 +
192 + // ── Session expiry handling ──
193 +
194 + #[tokio::test]
195 + async fn expired_token_detected_before_push() {
196 + let server = MockServer::start().await;
197 + let client = client_for(&server);
198 + let (user_id, app_id) = test_ids();
199 +
200 + let expired = fake_jwt(Utc::now().timestamp() - 100);
201 + client.restore_session(&expired, user_id, app_id);
202 + client.set_master_key_raw(synckit_client::crypto::generate_master_key());
203 +
204 + let err = client
205 + .push(DeviceId::new(Uuid::new_v4()), vec![])
206 + .await
207 + .unwrap_err();
208 + assert!(
209 + matches!(err, SyncKitError::TokenExpired),
210 + "Expired token should be detected pre-flight, got: {err:?}"
211 + );
212 + }
213 +
214 + #[tokio::test]
215 + async fn expired_token_detected_before_pull() {
216 + let server = MockServer::start().await;
217 + let client = client_for(&server);
218 + let (user_id, app_id) = test_ids();
219 +
220 + let expired = fake_jwt(Utc::now().timestamp() - 100);
221 + client.restore_session(&expired, user_id, app_id);
222 + client.set_master_key_raw(synckit_client::crypto::generate_master_key());
223 +
224 + let err = client
225 + .pull(DeviceId::new(Uuid::new_v4()), 0)
226 + .await
227 + .unwrap_err();
228 + assert!(matches!(err, SyncKitError::TokenExpired));
229 + }
230 +
231 + #[tokio::test]
232 + async fn expired_token_detected_before_register_device() {
233 + let server = MockServer::start().await;
234 + let client = client_for(&server);
235 + let (user_id, app_id) = test_ids();
236 +
237 + let expired = fake_jwt(Utc::now().timestamp() - 100);
238 + client.restore_session(&expired, user_id, app_id);
239 +
240 + let err = client.register_device("Test", "test").await.unwrap_err();
241 + assert!(matches!(err, SyncKitError::TokenExpired));
242 + }
243 +
244 + #[tokio::test]
245 + async fn expired_token_detected_before_list_devices() {
246 + let server = MockServer::start().await;
247 + let client = client_for(&server);
248 + let (user_id, app_id) = test_ids();
249 +
250 + let expired = fake_jwt(Utc::now().timestamp() - 100);
251 + client.restore_session(&expired, user_id, app_id);
252 +
253 + let err = client.list_devices().await.unwrap_err();
254 + assert!(matches!(err, SyncKitError::TokenExpired));
255 + }
256 +
257 + // ── Session edge cases ──
258 +
259 + #[tokio::test]
260 + async fn double_authenticate_overwrites_session() {
261 + let server = MockServer::start().await;
262 +
263 + let (user_id, app_id) = test_ids();
264 + let second_user_id = UserId::new(Uuid::new_v4());
265 +
266 + // First auth response
267 + Mock::given(method("POST"))
268 + .and(path("/api/v1/sync/auth"))
269 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
270 + "token": fresh_token(),
271 + "user_id": user_id,
272 + "app_id": app_id,
273 + })))
274 + .up_to_n_times(1)
275 + .mount(&server)
276 + .await;
277 +
278 + // Second auth response with different user_id
279 + Mock::given(method("POST"))
280 + .and(path("/api/v1/sync/auth"))
281 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
282 + "token": fresh_token(),
283 + "user_id": second_user_id,
284 + "app_id": app_id,
285 + })))
286 + .mount(&server)
287 + .await;
288 +
289 + let client = client_for(&server);
290 + let (uid1, _) = client
291 + .authenticate("user1@test.com", "pass1", "test-key")
292 + .await
293 + .unwrap();
294 + assert_eq!(uid1, user_id);
295 +
296 + let (uid2, _) = client
297 + .authenticate("user2@test.com", "pass2", "test-key")
298 + .await
299 + .unwrap();
300 + assert_eq!(uid2, second_user_id);
301 +
302 + // Session should now reflect the second auth
303 + let info = client.session_info().unwrap();
304 + assert_eq!(info.user_id, second_user_id);
305 + }
306 +
307 + #[tokio::test]
308 + async fn clear_session_then_authenticate_succeeds() {
309 + let server = MockServer::start().await;
310 +
311 + Mock::given(method("POST"))
312 + .and(path("/api/v1/sync/auth"))
313 + .respond_with(ResponseTemplate::new(200).set_body_json(auth_response_json()))
314 + .mount(&server)
315 + .await;
316 +
317 + let client = authed_client(&server);
318 + assert!(client.session_info().is_some());
319 +
320 + client.clear_session();
321 + assert!(client.session_info().is_none());
322 +
323 + // Re-authenticate should work
324 + let result = client
325 + .authenticate("user@test.com", "pass", "test-key")
326 + .await;
327 + assert!(
328 + result.is_ok(),
329 + "Should be able to re-authenticate after clear: {result:?}"
330 + );
331 + assert!(client.session_info().is_some());
332 + }
333 +
334 + #[tokio::test]
335 + async fn restore_session_with_expired_token_then_push_returns_token_expired() {
336 + let server = MockServer::start().await;
337 + let client = client_for(&server);
338 + let (user_id, app_id) = test_ids();
339 +
340 + let expired = fake_jwt(Utc::now().timestamp() - 3600);
341 + client.restore_session(&expired, user_id, app_id);
342 + client.set_master_key_raw(synckit_client::crypto::generate_master_key());
343 +
344 + let err = client
345 + .push(DeviceId::new(Uuid::new_v4()), vec![])
346 + .await
347 + .unwrap_err();
348 + assert!(
349 + matches!(err, SyncKitError::TokenExpired),
350 + "Restored expired token should return TokenExpired, got: {err:?}"
351 + );
352 + }
@@ -1,0 +1,469 @@
1 + //! Blob upload, download, and confirm over the one-shot PUT path, plus the size
2 + //! and key edge cases. The streaming path lives in [`blob_multipart`](super::blob_multipart).
3 +
4 + use crate::common::*;
5 +
6 + // ── Blob operations ──
7 +
8 + #[tokio::test]
9 + async fn blob_upload_url_success() {
10 + let server = MockServer::start().await;
11 +
12 + Mock::given(method("POST"))
13 + .and(path("/api/v1/sync/blobs/upload"))
14 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
15 + "upload_url": "https://s3.example.com/put",
16 + "already_exists": false,
17 + })))
18 + .mount(&server)
19 + .await;
20 +
21 + let client = authed_client(&server);
22 + let resp = client.blob_upload_url("sha256-abc", 1024).await.unwrap();
23 + assert_eq!(resp.upload_url, "https://s3.example.com/put");
24 + assert!(!resp.already_exists);
25 + }
26 +
27 + #[tokio::test]
28 + async fn blob_upload_url_declares_the_length_the_put_will_carry() {
29 + // The server signs the declared size into the presigned URL as
30 + // Content-Length, a SignedHeader, so declaring anything other than the
31 + // exact ciphertext length makes the PUT fail SigV4. The caller passes the
32 + // plaintext size it sees on disk; the SDK converts. This test pins the two
33 + // halves together, which is the only place the mismatch would show up:
34 + // wiremock does not verify signatures, and the server's own tests use an
35 + // in-memory backend that does not sign at all.
36 + let server = MockServer::start().await;
37 +
38 + let upload_path = "/s3/sized-upload";
39 + Mock::given(method("POST"))
40 + .and(path("/api/v1/sync/blobs/upload"))
41 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
42 + "upload_url": format!("{}{upload_path}", server.uri()),
43 + "already_exists": false,
44 + })))
45 + .mount(&server)
46 + .await;
47 + Mock::given(method("PUT"))
48 + .and(path(upload_path))
49 + .respond_with(ResponseTemplate::new(200))
50 + .mount(&server)
51 + .await;
52 +
53 + let client = authed_client(&server);
54 + client.set_master_key_raw(synckit_client::crypto::generate_master_key());
55 +
56 + // Spans two chunks, so the framing overhead is more than a single chunk's.
57 + let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE + 500))
58 + .map(|i| i as u8)
59 + .collect();
60 + let hash = hex::encode(sha2::Sha256::digest(&plaintext));
61 +
62 + let resp = client
63 + .blob_upload_url(&hash, plaintext.len() as i64)
64 + .await
65 + .unwrap();
66 + client
67 + .blob_upload(&hash, &resp.upload_url, plaintext.clone())
68 + .await
69 + .unwrap();
70 +
71 + let reqs = server.received_requests().await.unwrap();
72 + let declared: serde_json::Value = reqs
73 + .iter()
74 + .find(|r| r.url.path() == "/api/v1/sync/blobs/upload")
75 + .unwrap()
76 + .body_json()
77 + .unwrap();
78 + let put_len = reqs
79 + .iter()
80 + .find(|r| r.url.path() == upload_path)
81 + .unwrap()
82 + .body
83 + .len();
84 +
85 + assert_eq!(
86 + declared["size_bytes"].as_u64().unwrap(),
87 + put_len as u64,
88 + "the declared size must equal the bytes actually PUT, or the signature fails"
89 + );
90 + assert!(
91 + put_len > plaintext.len(),
92 + "the PUT carries ciphertext, which is longer than the plaintext"
93 + );
94 + }
95 +
96 + #[tokio::test]
97 + async fn blob_upload_encrypts_data() {
98 + let server = MockServer::start().await;
99 +
100 + let upload_path = "/s3/upload";
101 + Mock::given(method("PUT"))
102 + .and(path(upload_path))
103 + .respond_with(ResponseTemplate::new(200))
104 + .mount(&server)
105 + .await;
106 +
107 + let client = authed_client(&server);
108 + let key = synckit_client::crypto::generate_master_key();
109 + client.set_master_key_raw(key);
110 +
111 + let plaintext = b"hello blob data";
112 + let presigned = format!("{}{}", server.uri(), upload_path);
113 + client
114 + .blob_upload("sha256-test", &presigned, plaintext.to_vec())
115 + .await
116 + .unwrap();
117 +
118 + // Verify uploaded body is encrypted (not plaintext)
119 + let requests = server.received_requests().await.unwrap();
120 + let upload_req = requests
121 + .iter()
122 + .find(|r| r.url.path() == upload_path)
123 + .unwrap();
124 + assert!(
125 + !upload_req
126 + .body
127 + .windows(plaintext.len())
128 + .any(|w| w == plaintext),
129 + "Plaintext should not appear in uploaded body"
130 + );
131 + // Encrypted blob should be larger due to nonce + tag overhead
132 + assert!(upload_req.body.len() > plaintext.len());
133 + }
134 +
135 + #[tokio::test]
136 + async fn blob_download_decrypts_data() {
137 + let server = MockServer::start().await;
138 +
139 + let client = authed_client(&server);
140 + let key = synckit_client::crypto::generate_master_key();
141 + client.set_master_key_raw(key);
142 +
143 + // Encrypt data to simulate what S3 would return. A legacy (untagged) blob
144 + // still decrypts through the AAD-aware reader and must pass the hash check.
145 + let plaintext = b"decrypted blob content";
146 + let hash = hex::encode(sha2::Sha256::digest(plaintext));
147 + let encrypted = synckit_client::crypto::encrypt_bytes(plaintext, &key).unwrap();
148 +
149 + let download_path = "/s3/download";
150 + Mock::given(method("GET"))
151 + .and(path(download_path))
152 + .respond_with(ResponseTemplate::new(200).set_body_bytes(encrypted))
153 + .mount(&server)
154 + .await;
155 +
156 + let presigned = format!("{}{}", server.uri(), download_path);
157 + let result = client.blob_download(&hash, &presigned).await.unwrap();
158 + assert_eq!(result, plaintext);
159 + }
160 +
161 + #[tokio::test]
162 + async fn blob_upload_retries_on_503() {
163 + let server = MockServer::start().await;
164 +
165 + let upload_path = "/s3/retry-upload";
166 + Mock::given(method("PUT"))
167 + .and(path(upload_path))
168 + .respond_with(ResponseTemplate::new(503))
169 + .up_to_n_times(1)
170 + .mount(&server)
171 + .await;
172 +
173 + Mock::given(method("PUT"))
174 + .and(path(upload_path))
175 + .respond_with(ResponseTemplate::new(200))
176 + .mount(&server)
177 + .await;
178 +
179 + let client = authed_client(&server);
180 + let key = synckit_client::crypto::generate_master_key();
181 + client.set_master_key_raw(key);
182 +
183 + let presigned = format!("{}{}", server.uri(), upload_path);
184 + let result = client
185 + .blob_upload("sha256-x", &presigned, b"data".to_vec())
186 + .await;
187 + assert!(result.is_ok(), "Should succeed after retry: {result:?}");
188 + }
189 +
190 + // ── Blob confirm ──
191 +
192 + #[tokio::test]
193 + async fn blob_confirm_success() {
194 + let server = MockServer::start().await;
195 +
196 + Mock::given(method("POST"))
197 + .and(path("/api/v1/sync/blobs/confirm"))
198 + .respond_with(ResponseTemplate::new(200))
199 + .mount(&server)
200 + .await;
201 +
202 + let client = authed_client(&server);
203 + client.blob_confirm("sha256-abc", 1024).await.unwrap();
204 + }
205 +
206 + // ── Blob download URL ──
207 +
208 + #[tokio::test]
209 + async fn blob_download_url_success() {
210 + let server = MockServer::start().await;
211 +
212 + Mock::given(method("POST"))
213 + .and(path("/api/v1/sync/blobs/download"))
214 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
215 + "download_url": "https://s3.example.com/get",
216 + })))
217 + .mount(&server)
218 + .await;
219 +
220 + let client = authed_client(&server);
221 + let url = client.blob_download_url("sha256-abc").await.unwrap();
222 + assert_eq!(url, "https://s3.example.com/get");
223 + }
224 +
225 + // ── Blob edge cases ──
226 +
227 + #[tokio::test]
228 + async fn blob_upload_zero_byte_data() {
229 + let server = MockServer::start().await;
230 +
231 + let upload_path = "/s3/zero-byte";
232 + Mock::given(method("PUT"))
233 + .and(path(upload_path))
234 + .respond_with(ResponseTemplate::new(200))
235 + .mount(&server)
236 + .await;
237 +
238 + let client = authed_client(&server);
239 + let key = synckit_client::crypto::generate_master_key();
240 + client.set_master_key_raw(key);
241 +
242 + let presigned = format!("{}{}", server.uri(), upload_path);
243 + let result = client.blob_upload("sha256-empty", &presigned, vec![]).await;
244 + assert!(result.is_ok(), "Zero-byte blob upload should succeed");
245 +
246 + // Verify the uploaded data is the v3 chunked framing over an empty blob.
247 + let requests = server.received_requests().await.unwrap();
248 + let req = requests
249 + .iter()
250 + .find(|r| r.url.path() == upload_path)
251 + .unwrap();
252 + assert_eq!(
253 + req.body.len(),
254 + synckit_client::crypto::chunked_blob_overhead(0),
255 + "Empty plaintext should produce exactly the chunked overhead bytes"
256 + );
257 + }
258 +
259 + #[tokio::test]
260 + async fn blob_upload_download_roundtrip() {
261 + let server = MockServer::start().await;
262 +
263 + let client = authed_client(&server);
264 + let key = synckit_client::crypto::generate_master_key();
265 + client.set_master_key_raw(key);
266 +
267 + let plaintext = b"roundtrip blob data with special bytes \x00\xFF\x01";
268 + let hash = hex::encode(sha2::Sha256::digest(plaintext));
269 +
270 + // Upload
271 + let upload_path = "/s3/roundtrip-upload";
272 + Mock::given(method("PUT"))
273 + .and(path(upload_path))
274 + .respond_with(ResponseTemplate::new(200))
275 + .mount(&server)
276 + .await;
277 +
278 + client
279 + .blob_upload(
280 + &hash,
281 + &format!("{}{}", server.uri(), upload_path),
282 + plaintext.to_vec(),
283 + )
284 + .await
285 + .unwrap();
286 +
287 + // Capture what was uploaded
288 + let requests = server.received_requests().await.unwrap();
289 + let uploaded_body = &requests
290 + .iter()
291 + .find(|r| r.url.path() == upload_path)
292 + .unwrap()
293 + .body;
294 +
295 + // Serve that exact encrypted data back for download
296 + let download_path = "/s3/roundtrip-download";
297 + Mock::given(method("GET"))
298 + .and(path(download_path))
299 + .respond_with(ResponseTemplate::new(200).set_body_bytes(uploaded_body.clone()))
300 + .mount(&server)
301 + .await;
302 +
303 + let downloaded = client
304 + .blob_download(&hash, &format!("{}{}", server.uri(), download_path))
305 + .await
306 + .unwrap();
307 +
308 + assert_eq!(downloaded, plaintext, "Blob roundtrip must preserve data");
309 + }
310 +
311 + // ── Blob operations require auth ──
312 +
313 + #[tokio::test]
314 + async fn blob_upload_url_without_auth_fails() {
315 + let server = MockServer::start().await;
316 + let client = client_for(&server);
317 +
318 + let result = client.blob_upload_url("hash", 100).await;
319 + match result {
320 + Err(SyncKitError::NotAuthenticated) => {} // expected
321 + Err(other) => panic!("Expected NotAuthenticated, got: {other:?}"),
322 + Ok(_) => panic!("Expected NotAuthenticated error, got Ok"),
323 + }
324 + }
325 +
326 + #[tokio::test]
327 + async fn blob_confirm_without_auth_fails() {
328 + let server = MockServer::start().await;
329 + let client = client_for(&server);
330 +
331 + let err = client.blob_confirm("hash", 100).await.unwrap_err();
332 + assert!(matches!(err, SyncKitError::NotAuthenticated));
333 + }
334 +
335 + #[tokio::test]
336 + async fn blob_download_url_without_auth_fails() {
337 + let server = MockServer::start().await;
338 + let client = client_for(&server);
339 +
340 + let err = client.blob_download_url("hash").await.unwrap_err();
341 + assert!(matches!(err, SyncKitError::NotAuthenticated));
342 + }
343 +
344 + // ── Blob download with wrong key ──
345 +
346 + #[tokio::test]
347 + async fn blob_download_with_wrong_key_fails() {
348 + let server = MockServer::start().await;
349 +
350 + let key1 = synckit_client::crypto::generate_master_key();
351 + let key2 = synckit_client::crypto::generate_master_key();
352 +
353 + // Encrypt with key1
354 + let plaintext = b"encrypted with key1";
355 + let encrypted = synckit_client::crypto::encrypt_bytes(plaintext, &key1).unwrap();
356 +
357 + let download_path = "/s3/wrong-key";
358 + Mock::given(method("GET"))
359 + .and(path(download_path))
360 + .respond_with(ResponseTemplate::new(200).set_body_bytes(encrypted))
361 + .mount(&server)
362 + .await;
363 +
364 + // Client has key2 (wrong key)
365 + let client = authed_client(&server);
366 + client.set_master_key_raw(key2);
367 +
368 + let result = client
369 + .blob_download("sha256-x", &format!("{}{}", server.uri(), download_path))
370 + .await;
371 +
372 + assert!(
373 + result.is_err(),
374 + "Download with wrong key should fail: {result:?}"
375 + );
376 + assert!(matches!(
377 + result.unwrap_err(),
378 + SyncKitError::DecryptionFailed
379 + ));
380 + }
381 +
382 + // ── Blob edge cases ──
383 +
384 + #[tokio::test]
385 + async fn blob_confirm_retries_on_503() {
386 + let server = MockServer::start().await;
387 +
388 + Mock::given(method("POST"))
389 + .and(path("/api/v1/sync/blobs/confirm"))
390 + .respond_with(ResponseTemplate::new(503).set_body_string("Service Unavailable"))
391 + .up_to_n_times(1)
392 + .mount(&server)
393 + .await;
394 +
395 + Mock::given(method("POST"))
396 + .and(path("/api/v1/sync/blobs/confirm"))
397 + .respond_with(ResponseTemplate::new(200))
398 + .mount(&server)
399 + .await;
400 +
401 + let client = authed_client(&server);
402 + let result = client.blob_confirm("sha256-retry", 512).await;
403 + assert!(result.is_ok(), "Should succeed after retry: {result:?}");
404 + }
405 +
406 + #[tokio::test]
407 + async fn blob_download_retries_on_503() {
408 + let server = MockServer::start().await;
409 +
410 + let client = authed_client(&server);
411 + let key = synckit_client::crypto::generate_master_key();
412 + client.set_master_key_raw(key);
413 +
414 + let plaintext = b"retry download test";
415 + let hash = hex::encode(sha2::Sha256::digest(plaintext));
416 + let encrypted = synckit_client::crypto::encrypt_bytes(plaintext, &key).unwrap();
417 +
418 + let download_path = "/s3/retry-download";
419 + Mock::given(method("GET"))
420 + .and(path(download_path))
421 + .respond_with(ResponseTemplate::new(503))
422 + .up_to_n_times(1)
423 + .mount(&server)
424 + .await;
425 +
426 + Mock::given(method("GET"))
427 + .and(path(download_path))
428 + .respond_with(ResponseTemplate::new(200).set_body_bytes(encrypted))
429 + .mount(&server)
430 + .await;
431 +
432 + let presigned = format!("{}{}", server.uri(), download_path);
433 + let result = client.blob_download(&hash, &presigned).await.unwrap();
434 + assert_eq!(result, plaintext);
435 + }
436 +
437 + #[tokio::test]
438 + async fn blob_upload_1mb_with_correct_overhead() {
439 + let server = MockServer::start().await;
440 +
441 + let upload_path = "/s3/1mb-upload";
442 + Mock::given(method("PUT"))
443 + .and(path(upload_path))
444 + .respond_with(ResponseTemplate::new(200))
445 + .mount(&server)
446 + .await;
447 +
448 + let client = authed_client(&server);
449 + let key = synckit_client::crypto::generate_master_key();
450 + client.set_master_key_raw(key);
451 +
452 + let plaintext: Vec<u8> = (0..1_048_576u32).map(|i| (i % 256) as u8).collect();
453 + let presigned = format!("{}{}", server.uri(), upload_path);
454 + client
455 + .blob_upload("sha256-1mb", &presigned, plaintext.clone())
456 + .await
457 + .unwrap();
458 +
459 + let requests = server.received_requests().await.unwrap();
460 + let upload_req = requests
461 + .iter()
462 + .find(|r| r.url.path() == upload_path)
463 + .unwrap();
464 + assert_eq!(
465 + upload_req.body.len(),
466 + plaintext.len() + synckit_client::crypto::chunked_blob_overhead(plaintext.len()),
467 + "1MB upload should add exactly the v3 chunked overhead"
468 + );
469 + }
@@ -1,0 +1,435 @@
1 + //! Multipart (streaming) blob upload.
2 +
3 + // ── Multipart blob upload (streaming) ──
4 + //
5 + // The transport for blobs above the server's one-shot PUT ceiling. What matters
6 + // here is that the client never buffers the whole ciphertext yet still produces
7 + // a byte-exact v3 blob: it seals 1 MiB chunks as it reads the file and cuts the
8 + // sealed stream at the part boundaries the server signed, which it could only
9 + // pick because `blob_encrypted_len` predicts the ciphertext size up front.
10 + use crate::common::*;
11 + use std::path::PathBuf;
12 +
13 + const START_PATH: &str = "/api/v1/sync/blobs/multipart/start";
14 + const PARTS_PATH: &str = "/api/v1/sync/blobs/multipart/parts";
15 + const COMPLETE_PATH: &str = "/api/v1/sync/blobs/multipart/complete";
16 + const ABORT_PATH: &str = "/api/v1/sync/blobs/multipart/abort";
17 + const PART_PUT_PATH: &str = "/s3/part";
18 +
19 + fn temp_blob(name: &str, contents: &[u8]) -> PathBuf {
20 + use std::sync::atomic::{AtomicU64, Ordering};
21 + static N: AtomicU64 = AtomicU64::new(0);
22 + let mut p = std::env::temp_dir();
23 + p.push(format!(
24 + "synckit_mp_{}_{}_{name}",
25 + std::process::id(),
26 + N.fetch_add(1, Ordering::Relaxed)
27 + ));
28 + std::fs::write(&p, contents).unwrap();
29 + p
30 + }
31 +
32 + /// Stands in for the server's part-URL minting: answers whatever window the
33 + /// client asked for, rather than a fixed list, since the client requests one
34 + /// part at a time (it can only checksum a part it has already sealed).
35 + struct PartsResponder {
36 + cipher_len: usize,
37 + part_size: usize,
38 + base: String,
39 + }
40 +
41 + impl wiremock::Respond for PartsResponder {
42 + fn respond(&self, req: &wiremock::Request) -> ResponseTemplate {
43 + let body: serde_json::Value = serde_json::from_slice(&req.body).unwrap();
44 + let first = body["first_part"].as_u64().unwrap() as usize;
45 + let count = body["count"].as_u64().unwrap() as usize;
46 + let part_count = self.cipher_len.div_ceil(self.part_size);
47 + let last = (first + count - 1).min(part_count);
48 +
49 + let parts: Vec<serde_json::Value> = (first..=last)
50 + .map(|n| {
51 + let content_length = if n == part_count {
52 + self.cipher_len - self.part_size * (part_count - 1)
53 + } else {
54 + self.part_size
55 + };
56 + json!({
57 + "part_number": n,
58 + "content_length": content_length,
59 + "url": format!("{}{PART_PUT_PATH}?partNumber={n}", self.base),
60 + })
61 + })
62 + .collect();
63 + ResponseTemplate::new(200).set_body_json(json!({ "parts": parts }))
64 + }
65 + }
66 +
67 + /// Mount the whole session: start (with the given plan), part-URL minting,
68 + /// the PUT target, and complete.
69 + async fn mount_session(server: &MockServer, cipher_len: usize, part_size: usize) -> u32 {
70 + let part_count = cipher_len.div_ceil(part_size) as u32;
71 +
72 + Mock::given(method("POST"))
73 + .and(path(START_PATH))
74 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
75 + "upload_id": "test-upload-id",
76 + "part_size": part_size,
77 + "part_count": part_count,
78 + "already_exists": false,
79 + })))
80 + .mount(server)
81 + .await;
82 + Mock::given(method("POST"))
83 + .and(path(PARTS_PATH))
84 + .respond_with(PartsResponder {
85 + cipher_len,
86 + part_size,
87 + base: server.uri(),
88 + })
89 + .mount(server)
90 + .await;
91 + Mock::given(method("PUT"))
92 + .and(path(PART_PUT_PATH))
93 + .respond_with(ResponseTemplate::new(200).append_header("ETag", "\"part-etag\""))
94 + .mount(server)
95 + .await;
96 + Mock::given(method("POST"))
97 + .and(path(COMPLETE_PATH))
98 + .respond_with(ResponseTemplate::new(204))
99 + .mount(server)
100 + .await;
101 +
102 + part_count
103 + }
104 +
105 + fn hits(reqs: &[wiremock::Request], p: &str) -> usize {
106 + reqs.iter().filter(|r| r.url.path() == p).count()
107 + }
108 +
109 + #[tokio::test]
110 + async fn streaming_upload_tiles_the_parts_into_a_valid_blob() {
111 + let server = MockServer::start().await;
112 + let client = authed_client(&server);
113 + let key = synckit_client::crypto::generate_master_key();
114 + client.set_master_key_raw(key);
115 +
116 + // Spans four 1 MiB chunks (three full plus a remainder), so sealed
117 + // chunks straddle part boundaries rather than lining up with them.
118 + let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE * 3 + 7))
119 + .map(|i| i as u8)
120 + .collect();
121 + let hash = hex::encode(sha2::Sha256::digest(&plaintext));
122 + let file = temp_blob("big.bin", &plaintext);
123 +
124 + let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
125 + let part_size = 1024 * 1024;
126 + let part_count = mount_session(&server, cipher_len, part_size).await;
127 + assert!(part_count > 1, "the fixture must actually be multipart");
128 +
129 + client.blob_upload_streaming(&hash, &file).await.unwrap();
130 +
131 + let reqs = server.received_requests().await.unwrap();
132 +
133 + // The session was sized in ciphertext, predicted from the plaintext.
134 + let start: serde_json::Value = reqs
135 + .iter()
136 + .find(|r| r.url.path() == START_PATH)
137 + .unwrap()
138 + .body_json()
139 + .unwrap();
140 + assert_eq!(start["size_bytes"].as_u64().unwrap(), cipher_len as u64);
141 + assert_eq!(start["hash"].as_str().unwrap(), hash);
142 +
143 + // Every part carried exactly the length the server signed for it.
144 + let puts: Vec<&wiremock::Request> = reqs
145 + .iter()
146 + .filter(|r| r.url.path() == PART_PUT_PATH)
147 + .collect();
148 + assert_eq!(puts.len() as u32, part_count, "one PUT per planned part");
149 + for (i, put) in puts.iter().enumerate() {
150 + let expected = if i as u32 == part_count - 1 {
151 + cipher_len - part_size * (part_count as usize - 1)
152 + } else {
153 + part_size
154 + };
155 + assert_eq!(put.body.len(), expected, "part {} length", i + 1);
156 + }
157 +
158 + // Each part was requested with the SHA-256 of exactly the bytes that
159 + // part then carried, which is what S3 rehashes against at write time.
160 + // The pairing is what matters: a checksum bound to the wrong part is
161 + // worse than none, since it would reject a correct upload.
162 + let part_reqs: Vec<&wiremock::Request> =
163 + reqs.iter().filter(|r| r.url.path() == PARTS_PATH).collect();
164 + assert_eq!(
165 + part_reqs.len() as u32,
166 + part_count,
167 + "one URL request per part: a digest exists only once the part is sealed"
168 + );
169 + for (i, req) in part_reqs.iter().enumerate() {
170 + let body: serde_json::Value = req.body_json().unwrap();
171 + assert_eq!(body["first_part"].as_u64().unwrap(), i as u64 + 1);
172 + assert_eq!(body["count"].as_u64().unwrap(), 1);
173 + let declared = body["checksums"][0].as_str().unwrap();
174 + let expected =
175 + base64::engine::general_purpose::STANDARD.encode(sha2::Sha256::digest(&puts[i].body));
176 + assert_eq!(
177 + declared,
178 + expected,
179 + "part {} checksum must match its bytes",
180 + i + 1
181 + );
182 + // And the client must actually send it: it is a signed header, so
183 + // dropping it would fail SigV4 at S3.
184 + assert_eq!(
185 + puts[i]
186 + .headers
187 + .get("x-amz-checksum-sha256")
188 + .expect("the PUT must carry the checksum header")
189 + .to_str()
190 + .unwrap(),
191 + declared
192 + );
193 + }
194 +
195 + // The concatenated parts are a valid v3 blob for this content address:
196 + // proof that streaming produced the same wire format as the in-memory
197 + // encrypt, boundaries and all.
198 + let assembled: Vec<u8> = puts.iter().flat_map(|r| r.body.clone()).collect();
199 + assert_eq!(assembled.len(), cipher_len);
200 + let decrypted = synckit_client::crypto::decrypt_blob_chunked(&assembled, &key, &hash).unwrap();
201 + assert_eq!(decrypted, plaintext, "streamed blob must round-trip");
202 +
203 + // Complete named every part, in order, with the ETag S3 returned.
204 + let complete: serde_json::Value = reqs
205 + .iter()
206 + .find(|r| r.url.path() == COMPLETE_PATH)
207 + .unwrap()
208 + .body_json()
209 + .unwrap();
210 + let named = complete["parts"].as_array().unwrap();
211 + assert_eq!(named.len() as u32, part_count);
212 + for (i, part) in named.iter().enumerate() {
213 + assert_eq!(part["part_number"].as_u64().unwrap(), i as u64 + 1);
214 + assert_eq!(part["etag"].as_str().unwrap(), "\"part-etag\"");
215 + }
216 + assert_eq!(hits(&reqs, ABORT_PATH), 0, "a clean upload must not abort");
217 +
218 + std::fs::remove_file(&file).ok();
219 + }
220 +
221 + #[tokio::test]
222 + async fn streaming_upload_rejects_a_server_part_plan_that_lies_about_geometry() {
223 + let server = MockServer::start().await;
224 + let client = authed_client(&server);
225 + let key = synckit_client::crypto::generate_master_key();
226 + client.set_master_key_raw(key);
227 +
228 + // A blob that genuinely spans several 1 MiB parts.
229 + let plaintext: Vec<u8> = (0..(synckit_client::crypto::BLOB_CHUNK_SIZE * 3 + 7))
230 + .map(|i| i as u8)
231 + .collect();
232 + let hash = hex::encode(sha2::Sha256::digest(&plaintext));
233 + let file = temp_blob("liar.bin", &plaintext);
234 +
235 + // Hostile server: claims the whole multi-part blob fits in ONE part.
236 + // Trusting it would defeat the one-part-in-memory bound, so the client
237 + // must refuse before minting or PUTting anything.
238 + Mock::given(method("POST"))
239 + .and(path(START_PATH))
240 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
241 + "upload_id": "test-upload-id",
242 + "part_size": 1024 * 1024,
243 + "part_count": 1,
244 + "already_exists": false,
245 + })))
246 + .mount(&server)
247 + .await;
248 +
249 + let err = client
250 + .blob_upload_streaming(&hash, &file)
251 + .await
252 + .unwrap_err();
253 + assert!(
254 + matches!(err, SyncKitError::Internal(ref m) if m.contains("does not match")),
255 + "expected a geometry-mismatch rejection, got {err:?}"
256 + );
257 + let reqs = server.received_requests().await.unwrap();
258 + assert_eq!(
259 + hits(&reqs, PART_PUT_PATH),
260 + 0,
261 + "no part may be uploaded once the plan is rejected"
262 + );
263 +
264 + std::fs::remove_file(&file).ok();
265 + }
266 +
267 + #[tokio::test]
268 + async fn streaming_upload_handles_an_empty_file() {
269 + let server = MockServer::start().await;
270 + let client = authed_client(&server);
271 + let key = synckit_client::crypto::generate_master_key();
272 + client.set_master_key_raw(key);
273 +
274 + let hash = hex::encode(sha2::Sha256::digest(b""));
275 + let file = temp_blob("empty.bin", b"");
276 + let cipher_len = synckit_client::crypto::blob_encrypted_len(0);
277 + mount_session(&server, cipher_len, 1024 * 1024).await;
278 +
279 + client.blob_upload_streaming(&hash, &file).await.unwrap();
280 +
281 + let reqs = server.received_requests().await.unwrap();
282 + let put = reqs.iter().find(|r| r.url.path() == PART_PUT_PATH).unwrap();
283 + assert_eq!(
284 + put.body.len(),
285 + cipher_len,
286 + "one part carries the whole blob"
287 + );
288 + assert_eq!(
289 + synckit_client::crypto::decrypt_blob_chunked(&put.body, &key, &hash).unwrap(),
290 + Vec::<u8>::new(),
291 + "an empty blob is still an authenticated single chunk"
292 + );
293 +
294 + std::fs::remove_file(&file).ok();
295 + }
296 +
297 + #[tokio::test]
298 + async fn streaming_upload_skips_when_the_server_already_has_the_content() {
299 + let server = MockServer::start().await;
300 + let client = authed_client(&server);
301 + client.set_master_key_raw(synckit_client::crypto::generate_master_key());
302 +
303 + Mock::given(method("POST"))
304 + .and(path(START_PATH))
305 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
306 + "upload_id": "",
307 + "part_size": 0,
308 + "part_count": 0,
309 + "already_exists": true,
310 + })))
311 + .mount(&server)
312 + .await;
313 +
314 + let plaintext = b"content the server already holds";
315 + let hash = hex::encode(sha2::Sha256::digest(plaintext));
316 + let file = temp_blob("dedup.bin", plaintext);
317 +
318 + client.blob_upload_streaming(&hash, &file).await.unwrap();
319 +
320 + // Dedup must cost nothing: no file bytes read out to the wire, no
321 + // session to clean up.
322 + let reqs = server.received_requests().await.unwrap();
323 + assert_eq!(hits(&reqs, PART_PUT_PATH), 0, "dedup must not upload parts");
324 + assert_eq!(hits(&reqs, COMPLETE_PATH), 0);
325 + assert_eq!(hits(&reqs, ABORT_PATH), 0);
326 +
327 + std::fs::remove_file(&file).ok();
328 + }
329 +
330 + #[tokio::test]
331 + async fn streaming_upload_aborts_when_the_file_no_longer_matches_its_hash() {
332 + // The caller hashed the file in an earlier pass. If it changed since,
333 + // storing it under the stale content address would poison the address:
334 + // every later download would re-hash and reject it. Fail here instead,
335 + // and release the parts.
336 + let server = MockServer::start().await;
337 + let client = authed_client(&server);
338 + client.set_master_key_raw(synckit_client::crypto::generate_master_key());
339 +
340 + let plaintext = b"the bytes actually on disk";
341 + let stale_hash = hex::encode(sha2::Sha256::digest(b"what the caller hashed earlier"));
342 + let file = temp_blob("changed.bin", plaintext);
343 +
344 + let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
345 + mount_session(&server, cipher_len, 1024 * 1024).await;
346 + Mock::given(method("POST"))
347 + .and(path(ABORT_PATH))
348 + .respond_with(ResponseTemplate::new(204))
349 + .mount(&server)
350 + .await;
351 +
352 + let err = client
353 + .blob_upload_streaming(&stale_hash, &file)
354 + .await
355 + .expect_err("a hash mismatch must not be uploaded");
356 + assert!(
357 + matches!(err, SyncKitError::IntegrityFailed { .. }),
358 + "expected IntegrityFailed, got {err:?}"
359 + );
360 +
361 + let reqs = server.received_requests().await.unwrap();
362 + assert_eq!(
363 + hits(&reqs, COMPLETE_PATH),
364 + 0,
365 + "a mismatched blob must not be assembled"
366 + );
367 + assert_eq!(hits(&reqs, ABORT_PATH), 1, "the session must be released");
368 +
369 + std::fs::remove_file(&file).ok();
370 + }
371 +
372 + #[tokio::test]
373 + async fn streaming_upload_aborts_when_a_part_upload_fails() {
374 + // Parts already sent are billed until the session is aborted, so any
375 + // failure past `start` has to release it.
376 + let server = MockServer::start().await;
377 + let client = authed_client(&server);
378 + client.set_master_key_raw(synckit_client::crypto::generate_master_key());
379 +
380 + let plaintext = b"a blob whose part upload will fail";
381 + let hash = hex::encode(sha2::Sha256::digest(plaintext));
382 + let file = temp_blob("failing.bin", plaintext);
383 + let cipher_len = synckit_client::crypto::blob_encrypted_len(plaintext.len());
384 +
385 + Mock::given(method("POST"))
386 + .and(path(START_PATH))
387 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
388 + "upload_id": "test-upload-id",
389 + "part_size": cipher_len,
390 + "part_count": 1,
391 + "already_exists": false,
392 + })))
393 + .mount(&server)
394 + .await;
395 + Mock::given(method("POST"))
396 + .and(path(PARTS_PATH))
397 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
398 + "parts": [{
399 + "part_number": 1,
400 + "content_length": cipher_len,
401 + "url": format!("{}{PART_PUT_PATH}", server.uri()),
402 + }]
403 + })))
404 + .mount(&server)
405 + .await;
406 + Mock::given(method("PUT"))
407 + .and(path(PART_PUT_PATH))
408 + .respond_with(ResponseTemplate::new(403))
409 + .mount(&server)
410 + .await;
411 + Mock::given(method("POST"))
412 + .and(path(ABORT_PATH))
413 + .respond_with(ResponseTemplate::new(204))
414 + .mount(&server)
415 + .await;
416 +
417 + let err = client
418 + .blob_upload_streaming(&hash, &file)
419 + .await
420 + .unwrap_err();
421 + assert!(
422 + matches!(err, SyncKitError::Server { status: 403, .. }),
423 + "got {err:?}"
424 + );
425 +
426 + let reqs = server.received_requests().await.unwrap();
427 + assert_eq!(hits(&reqs, COMPLETE_PATH), 0);
428 + assert_eq!(
429 + hits(&reqs, ABORT_PATH),
430 + 1,
431 + "a failed transfer must release its parts"
432 + );
433 +
434 + std::fs::remove_file(&file).ok();
435 + }
@@ -1,0 +1,94 @@
1 + //! Shared fixtures for the integration suite: the wiremock server, a client
2 + //! wired to it, and the JSON bodies the SyncKit server would return.
3 + //!
4 + //! Every test module imports this with `use crate::common::*;`. A helper earns a
5 + //! place here when a second module wants it; a fixture only one module uses stays
6 + //! in that module.
7 +
8 + pub(crate) use base64::Engine as _;
9 + pub(crate) use chrono::Utc;
10 + pub(crate) use serde_json::json;
11 + pub(crate) use sha2::Digest as _;
12 + pub(crate) use std::sync::Arc;
13 + pub(crate) use std::time::Duration;
14 + pub(crate) use uuid::Uuid;
15 + pub(crate) use wiremock::matchers::{method, path};
16 + pub(crate) use wiremock::{Mock, MockServer, ResponseTemplate};
17 +
18 + pub(crate) use synckit_client::{
19 + AppId, ChangeEntry, ChangeOp, DeviceId, Hlc, SyncKitClient, SyncKitConfig, SyncKitError, UserId,
20 + };
21 +
22 + pub(crate) fn fake_jwt(exp: i64) -> String {
23 + let header =
24 + base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(r#"{"alg":"HS256","typ":"JWT"}"#);
25 + let payload = json!({
26 + "sub": "550e8400-e29b-41d4-a716-446655440000",
27 + "app": "6ba7b810-9dad-11d1-80b4-00c04fd430c8",
28 + "exp": exp,
29 + "iat": exp - 3600,
30 + });
31 + let payload_b64 =
32 + base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(payload.to_string().as_bytes());
33 + let sig = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(b"fake-signature");
34 + format!("{header}.{payload_b64}.{sig}")
35 + }
36 +
37 + pub(crate) fn fresh_token() -> String {
38 + fake_jwt(Utc::now().timestamp() + 3600)
39 + }
40 +
41 + pub(crate) fn test_ids() -> (UserId, AppId) {
42 + (
43 + UserId::new(Uuid::parse_str("550e8400-e29b-41d4-a716-446655440000").unwrap()),
44 + AppId::new(Uuid::parse_str("6ba7b810-9dad-11d1-80b4-00c04fd430c8").unwrap()),
45 + )
46 + }
47 +
48 + /// Install the rustls crypto provider once. reqwest is built `rustls-no-provider`,
49 + /// so a real consumer app installs one at startup (audiofiles installs ring); these
50 + /// tests have no such app, so they install ring themselves before building a client.
51 + pub(crate) fn ensure_crypto_provider() {
52 + static PROVIDER: std::sync::Once = std::sync::Once::new();
53 + PROVIDER.call_once(|| {
54 + // Err means a provider is already installed, which is the outcome we want.
55 + let _ = rustls::crypto::ring::default_provider().install_default();
56 + });
57 + }
58 +
59 + pub(crate) fn client_for(server: &MockServer) -> SyncKitClient {
60 + ensure_crypto_provider();
61 + SyncKitClient::new(SyncKitConfig {
62 + server_url: server.uri(),
63 + api_key: "test-api-key".to_string(),
64 + })
65 + }
66 +
67 + pub(crate) fn authed_client(server: &MockServer) -> SyncKitClient {
68 + let client = client_for(server);
69 + let (user_id, app_id) = test_ids();
70 + client.restore_session(&fresh_token(), user_id, app_id);
71 + client
72 + }
73 +
74 + pub(crate) fn auth_response_json() -> serde_json::Value {
75 + let (user_id, app_id) = test_ids();
76 + json!({
77 + "token": fresh_token(),
78 + "user_id": user_id,
79 + "app_id": app_id,
80 + })
81 + }
82 +
83 + pub(crate) fn device_json() -> serde_json::Value {
84 + let (user_id, app_id) = test_ids();
85 + json!({
86 + "id": Uuid::new_v4(),
87 + "app_id": app_id,
88 + "user_id": user_id,
89 + "device_name": "Test Device",
90 + "platform": "test",
91 + "last_seen_at": "2025-01-01T00:00:00Z",
92 + "created_at": "2025-01-01T00:00:00Z",
93 + })
94 + }
@@ -1,0 +1,248 @@
1 + //! Concurrent use of one client: interleaved push/pull, parallel reads of the
2 + //! session and key state, and the stress cases. These assert the absence of
3 + //! panics and data corruption, not a particular interleaving.
4 +
5 + use crate::common::*;
6 +
7 + // ── Concurrent access ──
8 +
9 + #[tokio::test]
10 + async fn concurrent_push_pull_no_panics() {
11 + let server = MockServer::start().await;
12 +
13 + Mock::given(method("POST"))
14 + .and(path("/api/v1/sync/push"))
15 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 1})))
16 + .mount(&server)
17 + .await;
18 +
19 + let device_id = DeviceId::new(Uuid::new_v4());
20 + Mock::given(method("POST"))
21 + .and(path("/api/v1/sync/pull"))
22 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
23 + "changes": [],
24 + "cursor": 0,
25 + "has_more": false,
26 + })))
27 + .mount(&server)
28 + .await;
29 +
30 + let client = Arc::new(authed_client(&server));
31 + let key = synckit_client::crypto::generate_master_key();
32 + client.set_master_key_raw(key);
33 +
34 + let mut handles = Vec::new();
35 + for _ in 0..4 {
36 + let c = Arc::clone(&client);
37 + let did = device_id;
38 + handles.push(tokio::spawn(async move {
39 + let _ = c.push(did, vec![]).await;
40 + let _ = c.pull(did, 0).await;
41 + }));
42 + }
43 +
44 + for h in handles {
45 + h.await.unwrap(); // No panics
46 + }
47 + }
48 +
49 + // ── Concurrent operations ──
50 +
51 + #[tokio::test]
52 + async fn concurrent_push_operations_no_data_corruption() {
53 + let server = MockServer::start().await;
54 +
55 + Mock::given(method("POST"))
56 + .and(path("/api/v1/sync/push"))
57 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 1})))
58 + .mount(&server)
59 + .await;
60 +
61 + let client = Arc::new(authed_client(&server));
62 + let key = synckit_client::crypto::generate_master_key();
63 + client.set_master_key_raw(key);
64 +
65 + let mut handles = Vec::new();
66 + for i in 0..8 {
67 + let c = Arc::clone(&client);
68 + handles.push(tokio::spawn(async move {
69 + let device_id = DeviceId::new(Uuid::new_v4());
70 + let entry = ChangeEntry {
71 + table: format!("table_{i}"),
72 + op: ChangeOp::Insert,
73 + row_id: format!("row_{i}"),
74 + timestamp: Utc::now(),
75 + hlc: Hlc::zero(DeviceId::nil()),
76 + data: Some(json!({"index": i})),
77 + extra: serde_json::Map::default(),
78 + };
79 + c.push(device_id, vec![entry]).await
80 + }));
81 + }
82 +
83 + for h in handles {
84 + let result = h.await.unwrap();
85 + assert!(result.is_ok(), "Concurrent push should succeed: {result:?}");
86 + }
87 + }
88 +
89 + #[tokio::test]
90 + async fn concurrent_push_and_pull_interleaved() {
91 + let server = MockServer::start().await;
92 + let device_id = DeviceId::new(Uuid::new_v4());
93 +
94 + Mock::given(method("POST"))
95 + .and(path("/api/v1/sync/push"))
96 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 10})))
97 + .mount(&server)
98 + .await;
99 +
100 + Mock::given(method("POST"))
101 + .and(path("/api/v1/sync/pull"))
102 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
103 + "changes": [],
104 + "cursor": 10,
105 + "has_more": false,
106 + })))
107 + .mount(&server)
108 + .await;
109 +
110 + let client = Arc::new(authed_client(&server));
111 + let key = synckit_client::crypto::generate_master_key();
112 + client.set_master_key_raw(key);
113 +
114 + let mut handles = Vec::new();
115 + for i in 0..4 {
116 + let c = Arc::clone(&client);
117 + let did = device_id;
118 + handles.push(tokio::spawn(async move {
119 + // Alternate push and pull
120 + if i % 2 == 0 {
121 + c.push(did, vec![]).await.map(|_| ())
122 + } else {
123 + c.pull(did, 0).await.map(|_| ())
124 + }
125 + }));
126 + }
127 +
128 + for h in handles {
129 + let result = h.await.unwrap();
130 + assert!(
131 + result.is_ok(),
132 + "Interleaved push/pull should succeed: {result:?}"
133 + );
134 + }
135 + }
136 +
137 + // ── Concurrency stress tests ──
138 +
139 + #[tokio::test]
140 + async fn concurrent_session_info_reads() {
141 + let server = MockServer::start().await;
142 + let client = Arc::new(authed_client(&server));
143 +
144 + let mut handles = Vec::new();
145 + for _ in 0..50 {
146 + let c = Arc::clone(&client);
147 + handles.push(tokio::spawn(async move { c.session_info() }));
148 + }
149 +
150 + for h in handles {
151 + let info = h.await.unwrap();
152 + assert!(
153 + info.is_some(),
154 + "All concurrent reads should see the session"
155 + );
156 + }
157 + }
158 +
159 + #[tokio::test]
160 + async fn concurrent_has_master_key_reads() {
161 + let server = MockServer::start().await;
162 + let client = Arc::new(authed_client(&server));
163 + client.set_master_key_raw(synckit_client::crypto::generate_master_key());
164 +
165 + let mut handles = Vec::new();
166 + for _ in 0..50 {
167 + let c = Arc::clone(&client);
168 + handles.push(tokio::spawn(async move { c.has_master_key() }));
169 + }
170 +
171 + for h in handles {
172 + let has_key = h.await.unwrap();
173 + assert!(has_key, "All concurrent reads should see the master key");
174 + }
175 + }
176 +
177 + #[tokio::test]
178 + async fn concurrent_status_checks() {
179 + let server = MockServer::start().await;
180 +
181 + Mock::given(method("GET"))
182 + .and(path("/api/v1/sync/status"))
183 + .respond_with(
184 + ResponseTemplate::new(200)
185 + .set_body_json(json!({"total_changes": 5, "latest_cursor": 3})),
186 + )
187 + .mount(&server)
188 + .await;
189 +
190 + let client = Arc::new(authed_client(&server));
191 +
192 + let mut handles = Vec::new();
193 + for _ in 0..20 {
194 + let c = Arc::clone(&client);
195 + handles.push(tokio::spawn(async move { c.status().await }));
196 + }
197 +
198 + for h in handles {
199 + let result = h.await.unwrap();
200 + assert!(
201 + result.is_ok(),
202 + "All concurrent status checks should succeed: {result:?}"
203 + );
204 + assert_eq!(result.unwrap().total_changes, 5);
205 + }
206 + }
207 +
208 + #[tokio::test]
209 + async fn concurrent_push_100_entries_each() {
210 + let server = MockServer::start().await;
211 +
212 + Mock::given(method("POST"))
213 + .and(path("/api/v1/sync/push"))
214 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 1})))
215 + .mount(&server)
216 + .await;
217 +
218 + let client = Arc::new(authed_client(&server));
219 + let key = synckit_client::crypto::generate_master_key();
220 + client.set_master_key_raw(key);
221 +
222 + let mut handles = Vec::new();
223 + for batch in 0..4 {
224 + let c = Arc::clone(&client);
225 + handles.push(tokio::spawn(async move {
226 + let changes: Vec<ChangeEntry> = (0..100)
227 + .map(|i| ChangeEntry {
228 + table: format!("batch_{batch}"),
229 + op: ChangeOp::Insert,
230 + row_id: format!("row_{i}"),
231 + timestamp: Utc::now(),
232 + hlc: Hlc::zero(DeviceId::nil()),
233 + data: Some(json!({"index": i})),
234 + extra: serde_json::Map::default(),
235 + })
236 + .collect();
237 + c.push(DeviceId::new(Uuid::new_v4()), changes).await
238 + }));
239 + }
240 +
241 + for h in handles {
242 + let result = h.await.unwrap();
243 + assert!(
244 + result.is_ok(),
245 + "Concurrent 100-entry push should succeed: {result:?}"
246 + );
247 + }
248 + }
@@ -1,0 +1,133 @@
1 + //! Device registration and listing, including the name edge cases (empty,
2 + //! unicode) and the unauthenticated rejection.
3 +
4 + use crate::common::*;
5 +
6 + // ── Device management ──
7 +
8 + #[tokio::test]
9 + async fn register_device_success() {
10 + let server = MockServer::start().await;
11 +
12 + Mock::given(method("POST"))
13 + .and(path("/api/v1/sync/devices"))
14 + .respond_with(ResponseTemplate::new(200).set_body_json(device_json()))
15 + .mount(&server)
16 + .await;
17 +
18 + let client = authed_client(&server);
19 + let device = client.register_device("MacBook", "macos").await.unwrap();
20 + assert_eq!(device.device_name, "Test Device");
21 + }
22 +
23 + #[tokio::test]
24 + async fn register_device_retries_on_transient() {
25 + let server = MockServer::start().await;
26 +
27 + Mock::given(method("POST"))
28 + .and(path("/api/v1/sync/devices"))
29 + .respond_with(ResponseTemplate::new(502).set_body_string("Bad Gateway"))
30 + .up_to_n_times(1)
31 + .mount(&server)
32 + .await;
33 +
34 + Mock::given(method("POST"))
35 + .and(path("/api/v1/sync/devices"))
36 + .respond_with(ResponseTemplate::new(200).set_body_json(device_json()))
37 + .mount(&server)
38 + .await;
39 +
40 + let client = authed_client(&server);
41 + let result = client.register_device("MacBook", "macos").await;
42 + assert!(result.is_ok(), "Should succeed after retry: {result:?}");
43 + }
44 +
45 + #[tokio::test]
46 + async fn list_devices_success() {
47 + let server = MockServer::start().await;
48 +
49 + Mock::given(method("GET"))
50 + .and(path("/api/v1/sync/devices"))
51 + .respond_with(ResponseTemplate::new(200).set_body_json(json!([device_json()])))
52 + .mount(&server)
53 + .await;
54 +
55 + let client = authed_client(&server);
56 + let devices = client.list_devices().await.unwrap();
57 + assert_eq!(devices.len(), 1);
58 + assert_eq!(devices[0].device_name, "Test Device");
59 + }
60 +
61 + // ── List devices without auth ──
62 +
63 + #[tokio::test]
64 + async fn list_devices_without_auth_returns_not_authenticated() {
65 + let server = MockServer::start().await;
66 + let client = client_for(&server);
67 +
68 + let err = client.list_devices().await.unwrap_err();
69 + assert!(matches!(err, SyncKitError::NotAuthenticated));
70 + }
71 +
72 + // ── Device management edge cases ──
73 +
74 + #[tokio::test]
75 + async fn register_device_with_empty_name() {
76 + let server = MockServer::start().await;
77 +
78 + Mock::given(method("POST"))
79 + .and(path("/api/v1/sync/devices"))
80 + .respond_with(ResponseTemplate::new(200).set_body_json(device_json()))
81 + .mount(&server)
82 + .await;
83 +
84 + let client = authed_client(&server);
85 + // Empty name should not panic; server may accept or reject
86 + let result = client.register_device("", "macos").await;
87 + assert!(result.is_ok(), "Empty device name should not panic");
88 + }
89 +
90 + #[tokio::test]
91 + async fn register_device_with_unicode_name() {
92 + let server = MockServer::start().await;
93 +
94 + let (user_id, app_id) = test_ids();
95 + Mock::given(method("POST"))
96 + .and(path("/api/v1/sync/devices"))
97 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
98 + "id": Uuid::new_v4(),
99 + "app_id": app_id,
100 + "user_id": user_id,
101 + "device_name": "\u{30DE}\u{30C3}\u{30AF}\u{30D6}\u{30C3}\u{30AF}",
102 + "platform": "macos",
103 + "last_seen_at": "2025-01-01T00:00:00Z",
104 + "created_at": "2025-01-01T00:00:00Z",
105 + })))
106 + .mount(&server)
107 + .await;
108 +
109 + let client = authed_client(&server);
110 + let device = client
111 + .register_device("\u{30DE}\u{30C3}\u{30AF}\u{30D6}\u{30C3}\u{30AF}", "macos")
112 + .await
113 + .unwrap();
114 + assert_eq!(
115 + device.device_name,
116 + "\u{30DE}\u{30C3}\u{30AF}\u{30D6}\u{30C3}\u{30AF}"
117 + );
118 + }
119 +
120 + #[tokio::test]
121 + async fn list_devices_empty_array() {
122 + let server = MockServer::start().await;
123 +
124 + Mock::given(method("GET"))
125 + .and(path("/api/v1/sync/devices"))
126 + .respond_with(ResponseTemplate::new(200).set_body_json(json!([])))
127 + .mount(&server)
128 + .await;
129 +
130 + let client = authed_client(&server);
131 + let devices = client.list_devices().await.unwrap();
132 + assert!(devices.is_empty());
133 + }
@@ -1,0 +1,537 @@
1 + //! Envelope setup, password change, and server-key presence.
2 + //!
3 + //! `setup_encryption_new`/`_existing` are how a device gets the master key, and
4 + //! `change_password` re-wraps it. These are the paths where a wrong answer costs
5 + //! the user their data, so the negative cases outnumber the positive ones.
6 +
7 + use crate::common::*;
8 +
9 + // ── Key management ──
10 +
11 + #[tokio::test]
12 + async fn has_server_key_true_on_200() {
13 + let server = MockServer::start().await;
14 +
15 + Mock::given(method("GET"))
16 + .and(path("/api/v1/sync/keys"))
17 + .respond_with(
18 + ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": "envelope-data"})),
19 + )
20 + .mount(&server)
21 + .await;
22 +
23 + let client = authed_client(&server);
24 + assert!(client.has_server_key().await.unwrap());
25 + }
26 +
27 + #[tokio::test]
28 + async fn has_server_key_false_on_404() {
29 + let server = MockServer::start().await;
30 +
31 + Mock::given(method("GET"))
32 + .and(path("/api/v1/sync/keys"))
33 + .respond_with(ResponseTemplate::new(404))
34 + .mount(&server)
35 + .await;
36 +
37 + let client = authed_client(&server);
38 + assert!(!client.has_server_key().await.unwrap());
39 + }
40 +
41 + #[tokio::test]
42 + async fn has_server_key_retries_on_500() {
43 + let server = MockServer::start().await;
44 +
45 + Mock::given(method("GET"))
46 + .and(path("/api/v1/sync/keys"))
47 + .respond_with(ResponseTemplate::new(500).set_body_string("Internal Server Error"))
48 + .up_to_n_times(1)
49 + .mount(&server)
50 + .await;
51 +
52 + Mock::given(method("GET"))
53 + .and(path("/api/v1/sync/keys"))
54 + .respond_with(
55 + ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": "envelope"})),
56 + )
57 + .mount(&server)
58 + .await;
59 +
60 + let client = authed_client(&server);
61 + assert!(client.has_server_key().await.unwrap());
62 + }
63 +
64 + // ── change_password: CRITICAL bug fix tests ──
65 +
66 + /// Helper: set up a server that serves a wrapped master key envelope.
67 + /// Returns (client, master_key, envelope_json).
68 + fn setup_change_password_test(
69 + server: &MockServer,
70 + password: &str,
71 + ) -> (SyncKitClient, [u8; 32], String) {
72 + let client = authed_client(server);
73 + let master_key = synckit_client::crypto::generate_master_key();
74 + let envelope = synckit_client::crypto::wrap_master_key(&master_key, password).unwrap();
75 +
76 + // Cache the master key in the client (simulating normal logged-in state)
77 + client.set_master_key_raw(master_key);
78 +
79 + (client, master_key, envelope)
80 + }
81 +
82 + #[tokio::test]
83 + async fn change_password_wrong_old_password_with_cached_key_fails() {
84 + let server = MockServer::start().await;
85 + let (client, _master_key, envelope) = setup_change_password_test(&server, "correct-old-pass");
86 +
87 + // Server returns the envelope on GET
88 + Mock::given(method("GET"))
89 + .and(path("/api/v1/sync/keys"))
90 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": envelope})))
91 + .mount(&server)
92 + .await;
93 +
94 + // Attempt to change password with wrong old password.
95 + // The key IS cached, but the old password must still be validated.
96 + let result = client.change_password("wrong-old-pass", "new-pass").await;
97 +
98 + assert!(
99 + result.is_err(),
100 + "change_password must fail when old_password is wrong, even with cached key"
101 + );
102 + assert!(
103 + matches!(result.unwrap_err(), SyncKitError::DecryptionFailed),
104 + "Should get DecryptionFailed for wrong old password"
105 + );
106 + }
107 +
108 + #[tokio::test]
109 + async fn change_password_correct_old_password_with_cached_key_succeeds() {
110 + let server = MockServer::start().await;
111 + let (client, master_key, envelope) = setup_change_password_test(&server, "correct-old-pass");
112 +
113 + // Server returns the envelope on GET
114 + Mock::given(method("GET"))
115 + .and(path("/api/v1/sync/keys"))
116 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": envelope})))
117 + .mount(&server)
118 + .await;
119 +
120 + // Server accepts the new envelope on PUT
121 + Mock::given(method("PUT"))
122 + .and(path("/api/v1/sync/keys"))
123 + .respond_with(ResponseTemplate::new(200))
124 + .mount(&server)
125 + .await;
126 +
127 + let result = client.change_password("correct-old-pass", "new-pass").await;
128 + assert!(
129 + result.is_ok(),
130 + "change_password should succeed with correct old password"
131 + );
132 +
133 + // Verify the PUT request was made (new envelope was uploaded)
134 + let requests = server.received_requests().await.unwrap();
135 + let put_requests: Vec<_> = requests
136 + .iter()
137 + .filter(|r| r.method.as_str() == "PUT" && r.url.path() == "/api/v1/sync/keys")
138 + .collect();
139 + assert_eq!(put_requests.len(), 1, "Should have sent exactly one PUT");
140 +
141 + // Verify the new envelope can be unwrapped with the new password
142 + let put_body: serde_json::Value = serde_json::from_slice(&put_requests[0].body).unwrap();
143 + let new_envelope = put_body["encrypted_key"].as_str().unwrap();
144 + let recovered = synckit_client::crypto::unwrap_master_key(new_envelope, "new-pass").unwrap();
145 + assert_eq!(
146 + recovered, master_key,
147 + "New envelope should unwrap to the same master key"
148 + );
149 + }
150 +
151 + #[tokio::test]
152 + async fn change_password_wrong_old_password_without_cached_key_fails() {
153 + let server = MockServer::start().await;
154 + let master_key = synckit_client::crypto::generate_master_key();
155 + let envelope =
156 + synckit_client::crypto::wrap_master_key(&master_key, "correct-old-pass").unwrap();
157 +
158 + let client = authed_client(&server);
159 + // Deliberately NOT setting master key: no cached key
160 +
161 + Mock::given(method("GET"))
162 + .and(path("/api/v1/sync/keys"))
163 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": envelope})))
164 + .mount(&server)
165 + .await;
166 +
167 + let result = client.change_password("wrong-old-pass", "new-pass").await;
168 +
169 + assert!(
170 + result.is_err(),
171 + "change_password must fail with wrong old password even without cached key"
172 + );
173 + assert!(matches!(
174 + result.unwrap_err(),
175 + SyncKitError::DecryptionFailed
176 + ));
177 + }
178 +
179 + #[tokio::test]
180 + async fn change_password_old_envelope_invalid_with_new_password() {
181 + let server = MockServer::start().await;
182 + let (client, _master_key, envelope) = setup_change_password_test(&server, "old-pass");
183 +
184 + Mock::given(method("GET"))
185 + .and(path("/api/v1/sync/keys"))
186 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": envelope})))
187 + .mount(&server)
188 + .await;
189 +
190 + Mock::given(method("PUT"))
191 + .and(path("/api/v1/sync/keys"))
192 + .respond_with(ResponseTemplate::new(200))
193 + .mount(&server)
194 + .await;
195 +
196 + client
197 + .change_password("old-pass", "new-pass")
198 + .await
199 + .unwrap();
200 +
201 + // Old password should NOT work on the new envelope
202 + let requests = server.received_requests().await.unwrap();
203 + let put_req = requests
204 + .iter()
205 + .find(|r| r.method.as_str() == "PUT" && r.url.path() == "/api/v1/sync/keys")
206 + .unwrap();
207 + let body: serde_json::Value = serde_json::from_slice(&put_req.body).unwrap();
208 + let new_envelope = body["encrypted_key"].as_str().unwrap();
209 +
210 + let result = synckit_client::crypto::unwrap_master_key(new_envelope, "old-pass");
211 + assert!(
212 + result.is_err(),
213 + "Old password must not work on the new envelope"
214 + );
215 + }
216 +
217 + // ── Encryption setup ──
218 +
219 + #[tokio::test]
220 + async fn setup_encryption_new_stores_key_and_uploads_envelope() {
221 + let server = MockServer::start().await;
222 +
223 + Mock::given(method("PUT"))
224 + .and(path("/api/v1/sync/keys"))
225 + .respond_with(ResponseTemplate::new(200))
226 + .expect(1)
227 + .mount(&server)
228 + .await;
229 +
230 + let client = authed_client(&server);
231 + assert!(!client.has_master_key());
232 +
233 + client.setup_encryption_new("test-password").await.unwrap();
234 +
235 + // Master key should now be in memory
236 + assert!(client.has_master_key());
237 +
238 + // Verify the PUT body contains a valid envelope unwrappable with the same password
239 + let requests = server.received_requests().await.unwrap();
240 + let put_req = requests
241 + .iter()
242 + .find(|r| r.method.as_str() == "PUT" && r.url.path() == "/api/v1/sync/keys")
243 + .unwrap();
244 + let body: serde_json::Value = serde_json::from_slice(&put_req.body).unwrap();
245 + let envelope_str = body["encrypted_key"].as_str().unwrap();
246 + let recovered =
247 + synckit_client::crypto::unwrap_master_key(envelope_str, "test-password").unwrap();
248 + assert_eq!(recovered.len(), 32);
249 + }
250 +
251 + #[tokio::test]
252 + async fn setup_encryption_new_without_auth_fails() {
253 + let server = MockServer::start().await;
254 + let client = client_for(&server);
255 +
256 + let err = client.setup_encryption_new("password").await.unwrap_err();
257 + assert!(matches!(err, SyncKitError::NotAuthenticated));
258 + }
259 +
260 + #[tokio::test]
261 + async fn setup_encryption_new_retries_on_server_error() {
262 + let server = MockServer::start().await;
263 +
264 + Mock::given(method("PUT"))
265 + .and(path("/api/v1/sync/keys"))
266 + .respond_with(ResponseTemplate::new(500).set_body_string("Internal Server Error"))
267 + .up_to_n_times(1)
268 + .mount(&server)
269 + .await;
270 +
271 + Mock::given(method("PUT"))
272 + .and(path("/api/v1/sync/keys"))
273 + .respond_with(ResponseTemplate::new(200))
274 + .mount(&server)
275 + .await;
276 +
277 + let client = authed_client(&server);
278 + let result = client.setup_encryption_new("password").await;
279 + assert!(result.is_ok(), "Should succeed after retry: {result:?}");
280 + assert!(client.has_master_key());
281 + }
282 +
283 + #[tokio::test]
284 + async fn setup_encryption_existing_recovers_key() {
285 + let server = MockServer::start().await;
286 +
287 + let master_key = synckit_client::crypto::generate_master_key();
288 + let envelope = synckit_client::crypto::wrap_master_key(&master_key, "my-password").unwrap();
289 +
290 + Mock::given(method("GET"))
291 + .and(path("/api/v1/sync/keys"))
292 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": envelope})))
293 + .mount(&server)
294 + .await;
295 +
296 + let client = authed_client(&server);
297 + assert!(!client.has_master_key());
298 +
299 + client
300 + .setup_encryption_existing("my-password")
301 + .await
302 + .unwrap();
303 +
304 + assert!(client.has_master_key());
305 + }
306 +
307 + #[tokio::test]
308 + async fn setup_encryption_existing_wrong_password_fails() {
309 + let server = MockServer::start().await;
310 +
311 + let master_key = synckit_client::crypto::generate_master_key();
312 + let envelope =
313 + synckit_client::crypto::wrap_master_key(&master_key, "correct-password").unwrap();
314 +
315 + Mock::given(method("GET"))
316 + .and(path("/api/v1/sync/keys"))
317 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": envelope})))
318 + .mount(&server)
319 + .await;
320 +
321 + let client = authed_client(&server);
322 + let err = client
323 + .setup_encryption_existing("wrong-password")
324 + .await
325 + .unwrap_err();
326 + assert!(
327 + matches!(err, SyncKitError::DecryptionFailed),
328 + "Wrong password should produce DecryptionFailed: {err:?}"
329 + );
330 + assert!(!client.has_master_key());
331 + }
332 +
333 + #[tokio::test]
334 + async fn setup_encryption_existing_without_auth_fails() {
335 + let server = MockServer::start().await;
336 + let client = client_for(&server);
337 +
338 + let err = client
339 + .setup_encryption_existing("password")
340 + .await
341 + .unwrap_err();
342 + assert!(matches!(err, SyncKitError::NotAuthenticated));
343 + }
344 +
345 + #[tokio::test]
346 + async fn setup_encryption_existing_retries_on_server_error() {
347 + let server = MockServer::start().await;
348 +
349 + let master_key = synckit_client::crypto::generate_master_key();
350 + let envelope = synckit_client::crypto::wrap_master_key(&master_key, "password").unwrap();
351 +
352 + Mock::given(method("GET"))
353 + .and(path("/api/v1/sync/keys"))
354 + .respond_with(ResponseTemplate::new(502).set_body_string("Bad Gateway"))
355 + .up_to_n_times(1)
356 + .mount(&server)
357 + .await;
358 +
359 + Mock::given(method("GET"))
360 + .and(path("/api/v1/sync/keys"))
361 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": envelope})))
362 + .mount(&server)
363 + .await;
364 +
365 + let client = authed_client(&server);
366 + let result = client.setup_encryption_existing("password").await;
367 + assert!(result.is_ok(), "Should succeed after retry: {result:?}");
368 + assert!(client.has_master_key());
369 + }
370 +
371 + #[tokio::test]
372 + async fn setup_encryption_existing_no_server_key_returns_error() {
373 + let server = MockServer::start().await;
374 +
375 + Mock::given(method("GET"))
376 + .and(path("/api/v1/sync/keys"))
377 + .respond_with(ResponseTemplate::new(404).set_body_string("Not Found"))
378 + .mount(&server)
379 + .await;
380 +
381 + let client = authed_client(&server);
382 + let err = client
383 + .setup_encryption_existing("password")
384 + .await
385 + .unwrap_err();
386 + assert!(
387 + matches!(err, SyncKitError::Server { status: 404, .. }),
388 + "Missing server key should produce 404 error: {err:?}"
389 + );
390 + }
391 +
392 + /// Two-device roundtrip: device 1 generates key via setup_encryption_new,
393 + /// device 2 recovers it via setup_encryption_existing. Data encrypted by
394 + /// device 1 must be decryptable by device 2.
395 + #[tokio::test]
396 + async fn encryption_setup_cross_device_roundtrip() {
397 + let server = MockServer::start().await;
398 +
399 + // Device 1: setup_encryption_new
400 + Mock::given(method("PUT"))
401 + .and(path("/api/v1/sync/keys"))
402 + .respond_with(ResponseTemplate::new(200))
403 + .mount(&server)
404 + .await;
405 +
406 + Mock::given(method("POST"))
407 + .and(path("/api/v1/sync/push"))
408 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({"cursor": 1})))
409 + .mount(&server)
410 + .await;
411 +
412 + let client1 = authed_client(&server);
413 + client1
414 + .setup_encryption_new("shared-password")
415 + .await
416 + .unwrap();
417 +
418 + // Push encrypted data from device 1
419 + let device_id = DeviceId::new(Uuid::new_v4());
420 + let original_data = json!({"title": "cross-device test", "secret": true});
421 + client1
422 + .push(
423 + device_id,
424 + vec![ChangeEntry {
425 + table: "tasks".into(),
426 + op: ChangeOp::Insert,
427 + row_id: "cross-r1".into(),
428 + timestamp: Utc::now(),
429 + hlc: Hlc::zero(DeviceId::nil()),
430 + data: Some(original_data.clone()),
431 + extra: serde_json::Map::default(),
432 + }],
433 + )
434 + .await
435 + .unwrap();
436 +
437 + // Capture the envelope and encrypted data
438 + let requests = server.received_requests().await.unwrap();
439 + let put_req = requests
440 + .iter()
441 + .find(|r| r.method.as_str() == "PUT" && r.url.path() == "/api/v1/sync/keys")
442 + .unwrap();
443 + let put_body: serde_json::Value = serde_json::from_slice(&put_req.body).unwrap();
444 + let envelope = put_body["encrypted_key"].as_str().unwrap().to_string();
445 +
446 + let push_req = requests
447 + .iter()
448 + .find(|r| r.url.path() == "/api/v1/sync/push")
449 + .unwrap();
450 + let push_body: serde_json::Value = serde_json::from_slice(&push_req.body).unwrap();
451 + let encrypted_data = push_body["changes"][0]["data"].clone();
452 +
453 + // Device 2: setup_encryption_existing with same password
454 + server.reset().await;
455 +
456 + Mock::given(method("GET"))
457 + .and(path("/api/v1/sync/keys"))
458 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({"encrypted_key": envelope})))
459 + .mount(&server)
460 + .await;
461 +
462 + Mock::given(method("POST"))
463 + .and(path("/api/v1/sync/pull"))
464 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
465 + "changes": [{
466 + "seq": 1,
467 + "device_id": device_id,
468 + "table": "tasks",
469 + "op": "INSERT",
470 + "row_id": "cross-r1",
471 + "timestamp": "2025-06-01T12:00:00Z",
472 + "data": encrypted_data,
473 + }],
474 + "cursor": 1,
475 + "has_more": false,
476 + })))
477 + .mount(&server)
478 + .await;
479 +
480 + let client2 = authed_client(&server);
481 + client2
482 + .setup_encryption_existing("shared-password")
483 + .await
484 + .unwrap();
485 +
486 + // Pull and decrypt with device 2's recovered key
487 + let (changes, _, _) = client2.pull(device_id, 0).await.unwrap();
488 + assert_eq!(changes.len(), 1);
489 + assert_eq!(
490 + changes[0].data.as_ref().unwrap(),
491 + &original_data,
492 + "Data encrypted by device 1 must be decryptable by device 2"
493 + );
494 + }
495 +
496 + // ── has_server_key without auth ──
497 +
498 + #[tokio::test]
499 + async fn has_server_key_without_auth_returns_not_authenticated() {
500 + let server = MockServer::start().await;
Lines truncated
@@ -1,0 +1,541 @@
1 + //! Group key rotation: the membership batch the client builds, and reading a
2 + //! pull that spans GCK generations.
3 +
4 + /// Rotation is the removal primitive: the batch an admin posts becomes the new
5 + /// membership, so the server drops anyone absent from it and re-keys in the same
6 + /// transaction. These tests pin the batch the client builds, because everything
7 + /// the server can enforce depends on the client getting that batch right.
8 + mod membership_batch {
9 + use crate::common::*;
10 + use synckit_client::{
11 + GroupId, IdentityKeypair, IdentityPublicKey, generate_group_key, open_gck_grant,
12 + seal_gck_to_member,
13 + };
14 + use wiremock::matchers::path_regex;
15 +
16 + /// A member we control both halves of, so a grant sealed to them can be
17 + /// opened and checked rather than merely counted.
18 + struct Member {
19 + user_id: UserId,
20 + keypair: IdentityKeypair,
21 + }
22 +
23 + impl Member {
24 + fn new() -> Self {
25 + Self {
26 + user_id: UserId::new(Uuid::new_v4()),
27 + keypair: IdentityKeypair::generate(),
28 + }
29 + }
30 +
31 + fn pubkey_json(&self) -> serde_json::Value {
32 + json!({
33 + "user_id": self.user_id,
34 + "pubkey": self.keypair.public_key().to_base64(),
35 + })
36 + }
37 + }
38 +
39 + /// The client whose master key seeds the admin identity, plus that identity.
40 + fn admin_client(server: &MockServer) -> (SyncKitClient, IdentityKeypair) {
41 + let client = authed_client(server);
42 + let master = synckit_client::crypto::generate_master_key();
43 + let identity = IdentityKeypair::from_master_key(&master);
44 + client.set_master_key_raw(master);
45 + (client, identity)
46 + }
47 +
48 + /// Mount the two reads a rotation makes: the admin's own grant (for the
49 + /// current generation) and the member pubkey list (the re-seal inputs).
50 + async fn mount_reads(
51 + server: &MockServer,
52 + group_id: GroupId,
53 + gck: &[u8; 32],
54 + admin: &IdentityKeypair,
55 + admin_id: UserId,
56 + version: i32,
57 + members: &[&Member],
58 + ) {
59 + let sealed = seal_gck_to_member(gck, &admin.public_key(), &group_id.to_string(), version)
60 + .expect("seal admin grant");
61 + Mock::given(method("GET"))
62 + .and(path(format!("/api/v1/sync/groups/{group_id}/grant")))
63 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
64 + "sealed_gck": sealed,
65 + "gck_version": version,
66 + })))
67 + .mount(server)
68 + .await;
69 +
70 + let mut pubkeys = vec![json!({
71 + "user_id": admin_id,
72 + "pubkey": admin.public_key().to_base64(),
73 + })];
74 + pubkeys.extend(members.iter().map(|m| m.pubkey_json()));
75 + Mock::given(method("GET"))
76 + .and(path(format!("/api/v1/sync/groups/{group_id}/pubkeys")))
77 + .respond_with(ResponseTemplate::new(200).set_body_json(pubkeys))
78 + .mount(server)
79 + .await;
80 + }
81 +
82 + async fn mount_rotate(server: &MockServer) {
83 + Mock::given(method("POST"))
84 + .and(path_regex(r"^/api/v1/sync/groups/[^/]+/rotate$"))
85 + .respond_with(ResponseTemplate::new(204))
86 + .mount(server)
87 + .await;
88 + }
89 +
90 + /// The body the client POSTed to `/rotate`.
91 + async fn posted_batch(server: &MockServer) -> serde_json::Value {
92 + let reqs = server.received_requests().await.expect("requests");
93 + let rotate = reqs
94 + .iter()
95 + .find(|r| r.url.path().ends_with("/rotate"))
96 + .expect("a rotate request was sent");
97 + serde_json::from_slice(&rotate.body).expect("rotate body is JSON")
98 + }
99 +
100 + #[tokio::test]
101 + async fn removing_a_member_rekeys_and_reseals_to_everyone_who_stays() {
102 + let server = MockServer::start().await;
103 + let (client, admin_identity) = admin_client(&server);
104 + let (admin_id, _) = test_ids();
105 + let group_id = GroupId::new(Uuid::new_v4());
106 + let old_gck = generate_group_key();
107 +
108 + let bob = Member::new();
109 + let carol = Member::new();
110 + mount_reads(
111 + &server,
112 + group_id,
113 + &old_gck,
114 + &admin_identity,
115 + admin_id,
116 + 7,
117 + &[&bob, &carol],
118 + )
119 + .await;
120 + mount_rotate(&server).await;
121 +
122 + client
123 + .remove_member(group_id, carol.user_id)
124 + .await
125 + .expect("remove member");
126 +
127 + let batch = posted_batch(&server).await;
128 + assert_eq!(
129 + batch["gck_version"], 8,
130 + "the generation must advance past the one our grant reports"
131 + );
132 +
133 + let grants = batch["grants"].as_array().expect("grants array");
134 + assert_eq!(grants.len(), 2, "admin and bob, not carol: {grants:?}");
135 + let recipients: Vec<&str> = grants
136 + .iter()
137 + .map(|g| g["user_id"].as_str().expect("user_id"))
138 + .collect();
139 + assert!(recipients.contains(&admin_id.to_string().as_str()));
140 + assert!(recipients.contains(&bob.user_id.to_string().as_str()));
141 + assert!(
142 + !recipients.contains(&carol.user_id.to_string().as_str()),
143 + "the removed member must not be re-granted"
144 + );
145 +
146 + // The grants are real seals of one new key, not placeholders: Bob's opens,
147 + // and what comes out is neither the old GCK nor something private to the
148 + // admin's copy.
149 + let bobs = grants
150 + .iter()
151 + .find(|g| g["user_id"].as_str() == Some(&bob.user_id.to_string()))
152 + .expect("bob's grant");
153 + let new_gck = open_gck_grant(
154 + bobs["sealed_gck"].as_str().expect("sealed_gck"),
155 + &bob.keypair,
156 + &group_id.to_string(),
157 + 8,
158 + )
159 + .expect("bob opens his grant");
160 + assert_ne!(new_gck, old_gck, "rotation must mint a fresh key");
161 +
162 + let admins = grants
163 + .iter()
164 + .find(|g| g["user_id"].as_str() == Some(&admin_id.to_string()))
165 + .expect("admin's grant");
166 + let admin_copy = open_gck_grant(
167 + admins["sealed_gck"].as_str().expect("sealed_gck"),
168 + &admin_identity,
169 + &group_id.to_string(),
170 + 8,
171 + )
172 + .expect("admin opens their own grant");
173 + assert_eq!(
174 + admin_copy, new_gck,
175 + "every member must be sealed the same new key"
176 + );
177 + }
178 +
179 + #[tokio::test]
180 + async fn a_grant_cannot_be_opened_by_the_member_it_was_not_sealed_to() {
181 + let server = MockServer::start().await;
182 + let (client, admin_identity) = admin_client(&server);
183 + let (admin_id, _) = test_ids();
184 + let group_id = GroupId::new(Uuid::new_v4());
185 +
186 + let bob = Member::new();
187 + let carol = Member::new();
188 + mount_reads(
189 + &server,
190 + group_id,
191 + &generate_group_key(),
192 + &admin_identity,
193 + admin_id,
194 + 1,
195 + &[&bob, &carol],
196 + )
197 + .await;
198 + mount_rotate(&server).await;
199 +
200 + client
201 + .rotate_group_key(group_id, &[])
202 + .await
203 + .expect("rotate without removing anyone");
204 +
205 + let batch = posted_batch(&server).await;
206 + let bobs = batch["grants"]
207 + .as_array()
208 + .expect("grants")
209 + .iter()
210 + .find(|g| g["user_id"].as_str() == Some(&bob.user_id.to_string()))
211 + .expect("bob's grant")["sealed_gck"]
212 + .as_str()
213 + .expect("sealed_gck")
214 + .to_string();
215 +
216 + assert!(
217 + open_gck_grant(&bobs, &carol.keypair, &group_id.to_string(), 2).is_err(),
218 + "a grant sealed to bob must not open under carol's key"
219 + );
220 + }
221 +
222 + #[tokio::test]
223 + async fn an_empty_removal_set_rekeys_without_dropping_anyone() {
224 + let server = MockServer::start().await;
225 + let (client, admin_identity) = admin_client(&server);
226 + let (admin_id, _) = test_ids();
227 + let group_id = GroupId::new(Uuid::new_v4());
228 +
229 + let bob = Member::new();
230 + mount_reads(
231 + &server,
232 + group_id,
233 + &generate_group_key(),
234 + &admin_identity,
235 + admin_id,
236 + 3,
237 + &[&bob],
238 + )
239 + .await;
240 + mount_rotate(&server).await;
241 +
242 + client
243 + .rotate_group_key(group_id, &[])
244 + .await
245 + .expect("rekey after a suspected compromise");
246 +
247 + let batch = posted_batch(&server).await;
248 + assert_eq!(batch["gck_version"], 4);
249 + assert_eq!(
250 + batch["grants"].as_array().expect("grants").len(),
251 + 2,
252 + "a bare re-key keeps the whole membership"
253 + );
254 + }
255 +
256 + #[tokio::test]
257 + async fn a_member_pubkey_the_client_cannot_parse_aborts_the_rotation() {
258 + let server = MockServer::start().await;
259 + let (client, admin_identity) = admin_client(&server);
260 + let (admin_id, _) = test_ids();
261 + let group_id = GroupId::new(Uuid::new_v4());
262 +
263 + let sealed = seal_gck_to_member(
264 + &generate_group_key(),
265 + &admin_identity.public_key(),
266 + &group_id.to_string(),
267 + 1,
268 + )
269 + .expect("seal admin grant");
270 + Mock::given(method("GET"))
271 + .and(path(format!("/api/v1/sync/groups/{group_id}/grant")))
272 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
273 + "sealed_gck": sealed,
274 + "gck_version": 1,
275 + })))
276 + .mount(&server)
277 + .await;
278 + Mock::given(method("GET"))
279 + .and(path(format!("/api/v1/sync/groups/{group_id}/pubkeys")))
280 + .respond_with(ResponseTemplate::new(200).set_body_json(json!([
281 + { "user_id": admin_id, "pubkey": admin_identity.public_key().to_base64() },
282 + { "user_id": Uuid::new_v4(), "pubkey": "not-a-key" },
283 + ])))
284 + .mount(&server)
285 + .await;
286 + mount_rotate(&server).await;
287 +
288 + client
289 + .rotate_group_key(group_id, &[])
290 + .await
291 + .expect_err("an unreadable member key must not produce a partial rotation");
292 +
293 + let reqs = server.received_requests().await.expect("requests");
294 + assert_eq!(
295 + reqs.iter()
296 + .filter(|r| r.url.path().ends_with("/rotate"))
297 + .count(),
298 + 0,
299 + "nothing may be posted when the batch could not be built in full"
300 + );
301 + }
302 +
303 + /// `IdentityPublicKey` round-trips through the wire form the pubkey list uses.
304 + /// If this ever stops holding, every rotation silently degrades to the error
305 + /// path above.
306 + #[test]
307 + fn member_pubkeys_round_trip_through_base64() {
308 + let identity = IdentityKeypair::generate();
309 + let encoded = identity.public_key().to_base64();
310 + let decoded = IdentityPublicKey::from_base64(&encoded).expect("round-trip");
311 + assert_eq!(decoded.as_bytes(), identity.public_key().as_bytes());
312 + }
313 + }
314 +
315 + /// The client half of history-survives-rotation: one pull can span GCK
316 + /// generations, and each entry is opened under the key it was sealed with.
317 + mod generations {
318 + use crate::common::*;
319 + use synckit_client::{
320 + ChangeEntry, GroupId, IdentityKeypair, generate_group_key, seal_gck_to_member,
321 + };
322 + use wiremock::matchers::{path_regex, query_param};
323 +
324 + fn change(row: &str, title: &str) -> ChangeEntry {
325 + ChangeEntry {
326 + table: "tasks".to_string(),
327 + op: ChangeOp::Insert,
328 + row_id: row.to_string(),
329 + timestamp: Utc::now(),
330 + hlc: Hlc::zero(DeviceId::nil()),
331 + data: Some(json!({ "title": title })),
332 + extra: serde_json::Map::default(),
333 + }
334 + }
335 +
336 + /// Push one change under `gck` and return the ciphertext the client produced,
337 + /// so it can be served straight back in a pull. Going through the real push
338 + /// path keeps the fixture honest: no test-local reimplementation of the AAD
339 + /// binding to drift from the one the client uses.
340 + async fn sealed_entry(
341 + client: &SyncKitClient,
342 + server: &MockServer,
343 + group_id: GroupId,
344 + gck: &[u8; 32],
345 + device: DeviceId,
346 + row: &str,
347 + title: &str,
348 + ) -> serde_json::Value {
349 + let before = server.received_requests().await.expect("requests").len();
350 + client
351 + .group_push(group_id, gck, device, vec![change(row, title)])
352 + .await
353 + .expect("group push");
354 + let reqs = server.received_requests().await.expect("requests");
355 + let pushed = reqs[before..]
356 + .iter()
357 + .find(|r| r.url.path().ends_with("/push"))
358 + .expect("a push was sent");
359 + let body: serde_json::Value = serde_json::from_slice(&pushed.body).expect("push body");
360 + body["changes"][0].clone()
361 + }
362 +
363 + #[tokio::test]
364 + async fn a_pull_spanning_two_generations_opens_each_under_its_own_key() {
365 + let server = MockServer::start().await;
366 + let client = authed_client(&server);
367 + let master = synckit_client::crypto::generate_master_key();
368 + let identity = IdentityKeypair::from_master_key(&master);
369 + client.set_master_key_raw(master);
370 +
371 + let group_id = GroupId::new(Uuid::new_v4());
372 + let device = DeviceId::new(Uuid::new_v4());
373 + let group_ref = group_id.to_string();
374 + let gck_v1 = generate_group_key();
375 + let gck_v2 = generate_group_key();
376 +
377 + Mock::given(method("POST"))
378 + .and(path_regex(r"^/api/v1/sync/groups/[^/]+/push$"))
379 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "cursor": 1 })))
380 + .mount(&server)
381 + .await;
382 +
383 + let old = sealed_entry(
384 + &client,
385 + &server,
386 + group_id,
387 + &gck_v1,
388 + device,
389 + "r-old",
390 + "before rotation",
391 + )
392 + .await;
393 + let new = sealed_entry(
394 + &client,
395 + &server,
396 + group_id,
397 + &gck_v2,
398 + device,
399 + "r-new",
400 + "after rotation",
401 + )
402 + .await;
403 +
404 + // Each generation's grant is fetched by version. Serving only these two
405 + // means a client that ignored the per-entry version and asked for one key
406 + // would still get an answer, and then fail to decrypt half the batch.
407 + for (version, gck) in [(1, &gck_v1), (2, &gck_v2)] {
408 + let sealed =
409 + seal_gck_to_member(gck, &identity.public_key(), &group_ref, version).expect("seal");
410 + Mock::given(method("GET"))
411 + .and(path(format!("/api/v1/sync/groups/{group_id}/grant")))
412 + .and(query_param("version", version.to_string()))
413 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
414 + "sealed_gck": sealed,
415 + "gck_version": version,
416 + })))
417 + .mount(&server)
418 + .await;
419 + }
420 +
421 + Mock::given(method("POST"))
422 + .and(path_regex(r"^/api/v1/sync/groups/[^/]+/pull$"))
423 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
424 + "changes": [
425 + {
426 + "seq": 1,
427 + "device_id": device,
428 + "table": old["table"],
429 + "op": old["op"],
430 + "row_id": old["row_id"],
431 + "timestamp": old["timestamp"],
432 + "data": old["data"],
433 + "gck_version": 1,
434 + },
435 + {
436 + "seq": 2,
437 + "device_id": device,
438 + "table": new["table"],
439 + "op": new["op"],
440 + "row_id": new["row_id"],
441 + "timestamp": new["timestamp"],
442 + "data": new["data"],
443 + "gck_version": 2,
444 + },
445 + ],
446 + "cursor": 2,
447 + "has_more": false,
448 + })))
449 + .mount(&server)
450 + .await;
451 +
452 + let (changes, cursor, has_more) = client
453 + .group_pull_rich(group_id, 2, device, 0)
454 + .await
455 + .expect("a pull spanning generations must succeed");
456 +
457 + assert_eq!(cursor, 2);
458 + assert!(!has_more);
459 + assert_eq!(changes.len(), 2);
460 + assert_eq!(
461 + changes[0].entry.data.as_ref().expect("old plaintext"),
462 + &json!({ "title": "before rotation" }),
463 + "the pre-rotation entry must open under generation 1"
464 + );
465 + assert_eq!(
466 + changes[1].entry.data.as_ref().expect("new plaintext"),
467 + &json!({ "title": "after rotation" }),
468 + "the post-rotation entry must open under generation 2"
469 + );
470 + }
471 +
472 + /// An entry with no generation is what a server predating per-generation
473 + /// grants returns. The caller's current generation is the fallback, so an old
474 + /// server keeps working rather than failing every pull.
475 + #[tokio::test]
476 + async fn an_entry_without_a_generation_falls_back_to_the_current_one() {
477 + let server = MockServer::start().await;
478 + let client = authed_client(&server);
479 + let master = synckit_client::crypto::generate_master_key();
480 + let identity = IdentityKeypair::from_master_key(&master);
481 + client.set_master_key_raw(master);
482 +
483 + let group_id = GroupId::new(Uuid::new_v4());
484 + let device = DeviceId::new(Uuid::new_v4());
485 + let gck = generate_group_key();
486 +
487 + Mock::given(method("POST"))
488 + .and(path_regex(r"^/api/v1/sync/groups/[^/]+/push$"))
489 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "cursor": 1 })))
490 + .mount(&server)
491 + .await;
492 + let entry = sealed_entry(
493 + &client,
494 + &server,
495 + group_id,
496 + &gck,
497 + device,
498 + "r-legacy",
499 + "legacy row",
500 + )
Lines truncated
@@ -1,0 +1,28 @@
1 + //! Integration tests against a wiremock stand-in for the MNW SyncKit server.
2 + //!
3 + //! One binary, one module per domain. Each module verifies the full HTTP
4 + //! round-trip for its surface: the request the client builds, the retry and error
5 + //! classification around it, and the encryption on either side.
6 + //!
7 + //! Shared fixtures live in [`common`]; a module reaches them with
8 + //! `use crate::common::*;`.
9 +
10 + mod common;
11 +
12 + mod auth;
13 + mod blob;
14 + mod blob_multipart;
15 + mod concurrency;
16 + mod device;
17 + mod encryption;
18 + mod group_rotation;
19 + // `rotate_key` finishes by caching the new key through `keystore::store_key`,
20 + // which hits the OS secret service under `keychain` and is unavailable on a
21 + // headless host. With the feature off it is the no-op stub, so the orchestration
22 + // runs hermetically. Run with:
23 + // cargo test --no-default-features --features store,testing
24 + #[cfg(not(feature = "keychain"))]
25 + mod rotation;
26 + mod subscribe;
27 + mod sync;
28 + mod transport;
@@ -1,0 +1,175 @@
1 + //! End-to-end master-key rotation against the full server protocol.
2 +
3 + // ── End-to-end key-rotation orchestration ──
4 + //
5 + // These drive `rotate_key()` through the full server protocol against wiremock:
6 + // fetch key -> begin -> re-encrypt loop -> complete, plus the straggler retry on
7 + // a 409. They are gated `#[cfg(not(feature = "keychain"))]` because `rotate_key`
8 + // finishes by caching the new key with `keystore::store_key`, which hits the OS
9 + // secret-service under the `keychain` feature and is unavailable on a headless
10 + // test host. With keychain off, `store_key` is the no-op stub, so the test
11 + // exercises the orchestration hermetically. Run with:
12 + // cargo test --no-default-features --features store,testing
13 + use crate::common::*;
14 +
15 + const KEYS_PATH: &str = "/api/v1/sync/keys";
16 + const ROTATE_PATH: &str = "/api/v1/sync/keys/rotate";
17 + const ENTRIES_PATH: &str = "/api/v1/sync/keys/rotate/entries";
18 + const BATCH_PATH: &str = "/api/v1/sync/keys/rotate/batch";
19 + const COMPLETE_PATH: &str = "/api/v1/sync/keys/rotate/complete";
20 +
21 + const ROTATE_PW: &str = "rotate-password";
22 +
23 + /// A `GET /keys` body wrapping `old_key` under [`ROTATE_PW`], with no rotation
24 + /// in progress, so `rotate_key` verifies the password and mints a fresh key.
25 + fn get_keys_body(old_key: &[u8; 32]) -> serde_json::Value {
26 + let envelope = synckit_client::crypto::wrap_master_key(old_key, ROTATE_PW).unwrap();
27 + json!({ "encrypted_key": envelope, "key_version": 1, "key_id": 1 })
28 + }
29 +
30 + /// One rotation entry: `plaintext` sealed under `old_key` with the same
31 + /// `(table, row_id)` AAD the client rebinds during re-encryption.
32 + fn rotation_entry(old_key: &[u8; 32], table: &str, row_id: &str) -> serde_json::Value {
33 + let ctx = synckit_client::crypto::AeadContext::entry(table, row_id);
34 + let sealed =
35 + synckit_client::crypto::encrypt_json_aad(&json!({"title": "secret"}), old_key, &ctx)
36 + .unwrap();
37 + json!({ "seq": 1, "table": table, "row_id": row_id, "data": sealed })
38 + }
39 +
40 + fn hits(reqs: &[wiremock::Request], p: &str) -> usize {
41 + reqs.iter().filter(|r| r.url.path() == p).count()
42 + }
43 +
44 + #[tokio::test]
45 + async fn rotate_key_drives_full_orchestration() {
46 + let server = MockServer::start().await;
47 + let old_key = synckit_client::crypto::generate_master_key();
48 +
49 + Mock::given(method("GET"))
50 + .and(path(KEYS_PATH))
51 + .respond_with(ResponseTemplate::new(200).set_body_json(get_keys_body(&old_key)))
52 + .mount(&server)
53 + .await;
54 + Mock::given(method("POST"))
55 + .and(path(ROTATE_PATH))
56 + .respond_with(ResponseTemplate::new(200).set_body_json(
57 + json!({ "rotation_id": Uuid::new_v4(), "target_seq": 1, "new_key_id": 2 }),
58 + ))
59 + .mount(&server)
60 + .await;
61 + // One batch of work, then drained (has_more = false ends the re-encrypt loop).
62 + Mock::given(method("POST"))
63 + .and(path(ENTRIES_PATH))
64 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
65 + "entries": [rotation_entry(&old_key, "tasks", "r1")],
66 + "has_more": false
67 + })))
68 + .mount(&server)
69 + .await;
70 + Mock::given(method("POST"))
71 + .and(path(BATCH_PATH))
72 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "updated_count": 1 })))
73 + .mount(&server)
74 + .await;
75 + Mock::given(method("POST"))
76 + .and(path(COMPLETE_PATH))
77 + .respond_with(ResponseTemplate::new(200))
78 + .mount(&server)
79 + .await;
80 +
81 + let client = authed_client(&server);
82 + client
83 + .rotate_key(DeviceId::new(Uuid::new_v4()), ROTATE_PW)
84 + .await
85 + .expect("full rotation should complete");
86 +
87 + // Every stage of the protocol was driven, in the right shape.
88 + let reqs = server.received_requests().await.unwrap();
89 + assert_eq!(hits(&reqs, KEYS_PATH), 1, "fetched key state once");
90 + assert_eq!(hits(&reqs, ROTATE_PATH), 1, "began rotation once");
91 + assert!(
92 + hits(&reqs, ENTRIES_PATH) >= 1,
93 + "pulled entries to re-encrypt"
94 + );
95 + assert_eq!(hits(&reqs, BATCH_PATH), 1, "pushed one re-encrypted batch");
96 + assert_eq!(
97 + hits(&reqs, COMPLETE_PATH),
98 + 1,
99 + "completed once (no stragglers)"
100 + );
101 + }
102 +
103 + #[tokio::test]
104 + async fn rotate_key_retries_reencrypt_on_straggler_conflict() {
105 + let server = MockServer::start().await;
106 + let old_key = synckit_client::crypto::generate_master_key();
107 +
108 + Mock::given(method("GET"))
109 + .and(path(KEYS_PATH))
110 + .respond_with(ResponseTemplate::new(200).set_body_json(get_keys_body(&old_key)))
111 + .mount(&server)
112 + .await;
113 + Mock::given(method("POST"))
114 + .and(path(ROTATE_PATH))
115 + .respond_with(ResponseTemplate::new(200).set_body_json(
116 + json!({ "rotation_id": Uuid::new_v4(), "target_seq": 1, "new_key_id": 2 }),
117 + ))
118 + .mount(&server)
119 + .await;
120 + // First entries pull returns work; every later pull is drained. Mounted in
121 + // this order so the up_to_n_times(1) mock wins the first call, then the
122 + // empty-set fallback serves the straggler round's re-pull.
123 + Mock::given(method("POST"))
124 + .and(path(ENTRIES_PATH))
125 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({
126 + "entries": [rotation_entry(&old_key, "tasks", "r1")],
127 + "has_more": false
128 + })))
129 + .up_to_n_times(1)
130 + .mount(&server)
131 + .await;
132 + Mock::given(method("POST"))
133 + .and(path(ENTRIES_PATH))
134 + .respond_with(
135 + ResponseTemplate::new(200).set_body_json(json!({ "entries": [], "has_more": false })),
136 + )
137 + .mount(&server)
138 + .await;
139 + Mock::given(method("POST"))
140 + .and(path(BATCH_PATH))
141 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "updated_count": 1 })))
142 + .mount(&server)
143 + .await;
144 + // First completion reports a straggler (409); the retry then succeeds.
145 + Mock::given(method("POST"))
146 + .and(path(COMPLETE_PATH))
147 + .respond_with(ResponseTemplate::new(409).set_body_json(json!({ "message": "stragglers" })))
148 + .up_to_n_times(1)
149 + .mount(&server)
150 + .await;
151 + Mock::given(method("POST"))
152 + .and(path(COMPLETE_PATH))
153 + .respond_with(ResponseTemplate::new(200))
154 + .mount(&server)
155 + .await;
156 +
157 + let client = authed_client(&server);
158 + client
159 + .rotate_key(DeviceId::new(Uuid::new_v4()), ROTATE_PW)
160 + .await
161 + .expect("rotation should converge after the straggler retry");
162 +
163 + // The 409 forced a second completion attempt, and the straggler round
164 + // re-ran the re-encrypt loop (a second entries pull).
165 + let reqs = server.received_requests().await.unwrap();
166 + assert_eq!(
167 + hits(&reqs, COMPLETE_PATH),
168 + 2,
169 + "completed twice: 409 then 200"
170 + );
171 + assert!(
172 + hits(&reqs, ENTRIES_PATH) >= 2,
173 + "straggler round re-pulled entries"
174 + );
175 + }
@@ -1,0 +1,105 @@
1 + //! The SSE subscribe stream and its reconnect state machine.
2 +
3 + use crate::common::*;
4 +
5 + // ── SSE subscribe / reconnect state machine ──
6 + //
7 + // These drive the real `subscribe()` -> `SyncNotifyStream::next_change()` path
8 + // against wiremock so the reconnect state machine (subscribe.rs) is exercised
9 + // end to end: a live notification, a transparent reconnect across a stream
10 + // drop, the fatal auth-rejection exit, and the give-up-after-N-failures cap.
11 +
12 + const SUBSCRIBE_PATH: &str = "/api/v1/sync/subscribe";
13 +
14 + /// One complete SSE "changed" block, body-terminated so the client sees a full
15 + /// event and then end-of-stream.
16 + fn sse_changed_block() -> &'static str {
17 + "event: changed\n\n"
18 + }
19 +
20 + #[tokio::test]
21 + async fn subscribe_yields_changed_event() {
22 + let server = MockServer::start().await;
23 + Mock::given(method("GET"))
24 + .and(path(SUBSCRIBE_PATH))
25 + .respond_with(ResponseTemplate::new(200).set_body_string(sse_changed_block()))
26 + .mount(&server)
27 + .await;
28 +
29 + let client = authed_client(&server);
30 + let mut stream = client.subscribe().await.expect("subscribe should succeed");
31 +
32 + // The first block is delivered inside the initial response body, so this
33 + // returns without any reconnect.
34 + assert_eq!(stream.next_change().await, Some(()));
35 + }
36 +
37 + #[tokio::test]
38 + async fn subscribe_reconnects_transparently_after_stream_drop() {
39 + let server = MockServer::start().await;
40 + // Every connection serves one block then ends (Content-Length terminates the
41 + // body). Consuming the first event, then reading past it, drops the stream
42 + // and forces a reconnect that must transparently yield the next event.
43 + Mock::given(method("GET"))
44 + .and(path(SUBSCRIBE_PATH))
45 + .respond_with(ResponseTemplate::new(200).set_body_string(sse_changed_block()))
46 + .mount(&server)
47 + .await;
48 +
49 + let client = authed_client(&server);
50 + let mut stream = client.subscribe().await.expect("subscribe should succeed");
51 +
52 + // #1 comes from the initial connection; #2 can only arrive after the stream
53 + // drops (body EOF) and reconnect() re-establishes it.
54 + assert_eq!(stream.next_change().await, Some(()));
55 + assert_eq!(stream.next_change().await, Some(()));
56 +
57 + // subscribe() opened one connection; the second event required at least one
58 + // reconnect, so the server saw two or more subscribe requests.
59 + let hits = server
60 + .received_requests()
61 + .await
62 + .unwrap()
63 + .into_iter()
64 + .filter(|r| r.url.path() == SUBSCRIBE_PATH)
65 + .count();
66 + assert!(
67 + hits >= 2,
68 + "expected a reconnect (>=2 subscribe requests), got {hits}"
69 + );
70 + }
71 +
72 + #[tokio::test]
73 + async fn subscribe_stream_closes_on_auth_rejection() {
74 + let server = MockServer::start().await;
75 + // First connection opens cleanly but carries no event and ends immediately,
76 + // forcing a reconnect. The reconnect is rejected for auth -> fatal, so the
77 + // stream ends with `None` rather than retrying.
78 + Mock::given(method("GET"))
79 + .and(path(SUBSCRIBE_PATH))
80 + .respond_with(ResponseTemplate::new(200).set_body_string(""))
81 + .up_to_n_times(1)
82 + .mount(&server)
83 + .await;
84 + Mock::given(method("GET"))
85 + .and(path(SUBSCRIBE_PATH))
86 + .respond_with(ResponseTemplate::new(401).set_body_string("unauthorized"))
87 + .mount(&server)
88 + .await;
89 +
90 + let client = authed_client(&server);
91 + let mut stream = client.subscribe().await.expect("initial subscribe is 200");
92 +
93 + // Empty body -> EOF -> reconnect -> 401 -> fatal -> None.
94 + assert_eq!(stream.next_change().await, None);
95 + }
96 +
97 + // Note: the "give up after MAX_RECONNECT_ATTEMPTS consecutive failures" path is
98 + // deliberately not covered end-to-end here. Exercising it against a live mock
99 + // server would incur the real exponential backoff (growing to a 60s cap, minutes
100 + // of wall-clock), and `tokio::time::pause()` cannot collapse it: the wiremock
101 + // server runs on its own runtime, so a paused clock auto-advances past the
102 + // cross-thread request to the nearest timer and the reconnect never completes.
103 + // The two pieces of that path are covered separately: the backoff schedule by
104 + // `reconnect_delay_grows_then_caps` (subscribe.rs), and the fatal-exit mechanics
105 + // (returning `None` and clearing state) by the auth-rejection test above.