Skip to main content

max / makenotwork

synckit: cap response bodies, harden TLS/loopback + transport (risk Phase 2) Drive the Transport/Resilience axis to A+ (crate-internal; no API/consumer break). - D1 (structural): one read_body_capped choke point (8 MiB control cap, streamed with a running total + Content-Length fast-reject). Route check_response, retry_request_json, and all 6 hand-rolled reads through it — including the public, unauthenticated OTA updater check that could OOM any installed app. clippy.toml disallowed-methods bans Response::json/text/bytes so a new uncapped read cannot compile. Integration test proves a 9 MiB body is rejected. - Loopback TLS: replace the starts_with prefix match with an exact authority parse (strip userinfo + port, IPv6 brackets) + IpAddr::is_loopback(), closing 127.0.0.1.attacker.com / localhost.evil.com / user@host cleartext-token bypasses. - http_stream: 90s read (inactivity) timeout above the server's 30s SSE keepalive, closing slow-loris on blob/OTA/SSE without capping long transfers. - rotate_key: bound the re-encrypt and straggler loops (MAX_ROTATION_ROUNDS) so a stalling/409-looping server can't spin the client forever. - D2 (structural): Idempotency::Safe -> ReadOnly | IdempotentWrite { on } so a mutating create auto-retried without naming its dedup key won't compile; the OTA creates now name the verified UNIQUE constraints (migration 033). Behavior preserved; basis logged on retry. - SSE reconnect gains a consecutive-failure ceiling; OTA URL construction moves into Endpoints. Gate: 308 lib + 101 integration + 1 doc tests, clippy --all-targets clean. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-07 18:55 UTC
Commit: ae1d01b1e155f5beba20100d709b34ba53f708a6
Parent: d5d4f2d
13 files changed, +333 insertions, -94 deletions
M shared/synckit-client/Cargo.lock +1 -1
@@ -1520,7 +1520,7 @@ dependencies = [
1520 1520
1521 1521 [[package]]
1522 1522 name = "synckit-client"
1523 - version = "0.5.0"
1523 + version = "0.6.0"
1524 1524 dependencies = [
1525 1525 "argon2",
1526 1526 "base64",
A shared/synckit-client/clippy.toml +13
@@ -0,0 +1,13 @@
1 + # A raw `Response::json`/`text`/`bytes` reads the whole body with no size limit,
2 + # so a hostile or buggy server can stream an arbitrarily large body and OOM the
3 + # client (worst case: the public, unauthenticated OTA updater check). Every body
4 + # read must go through `client::helpers::read_body_capped` (or its `read_json_capped`
5 + # / `read_text_capped` wrappers), which drains `bytes_stream()` under a hard cap.
6 + # `blob_download` streams `bytes_stream()` directly under its own 4 GiB cap.
7 + # This seal makes a new uncapped read a compile-time (clippy) error, not a latent
8 + # OOM lever someone has to remember to avoid.
9 + disallowed-methods = [
10 + { path = "reqwest::Response::json", reason = "unbounded body read; use client::helpers::read_json_capped" },
11 + { path = "reqwest::Response::text", reason = "unbounded body read; use client::helpers::read_text_capped" },
12 + { path = "reqwest::Response::bytes", reason = "unbounded body read; use client::helpers::read_body_capped (or bytes_stream() under an explicit cap)" },
13 + ]
M shared/synckit-client/src/client/auth.rs +3 -3
@@ -41,7 +41,7 @@ impl SyncKitClient {
41 41 })?);
42 42
43 43 let resp = self
44 - .retry_request(Idempotency::Safe, || {
44 + .retry_request(Idempotency::IdempotentWrite { on: "credentials; re-auth reissues an equivalent token" }, || {
45 45 let req = self
46 46 .http
47 47 .post(&self.endpoints.auth)
@@ -50,7 +50,7 @@ impl SyncKitClient {
50 50 async move { check_response(req.send().await?).await }
51 51 })
52 52 .await?;
53 - let auth: AuthResponse = resp.json().await?;
53 + let auth: AuthResponse = crate::client::helpers::read_json_capped(resp, crate::client::helpers::MAX_CONTROL_BODY_BYTES).await?;
54 54
55 55 let user_id = auth.user_id;
56 56 let app_id = auth.app_id;
@@ -177,7 +177,7 @@ impl SyncKitClient {
177 177 .await?,
178 178 )
179 179 .await?;
180 - let token_resp: OAuthTokenResponse = resp.json().await?;
180 + let token_resp: OAuthTokenResponse = crate::client::helpers::read_json_capped(resp, crate::client::helpers::MAX_CONTROL_BODY_BYTES).await?;
181 181
182 182 let user_id = token_resp.user_id;
183 183 let app_id = token_resp.app_id;
M shared/synckit-client/src/client/blob.rs +5 -5
@@ -39,7 +39,7 @@ impl SyncKitClient {
39 39 size_bytes,
40 40 })?);
41 41
42 - self.retry_request_json(Idempotency::Safe, || {
42 + self.retry_request_json(Idempotency::ReadOnly, || {
43 43 let req = self
44 44 .http
45 45 .post(&self.endpoints.blobs_upload)
@@ -70,7 +70,7 @@ impl SyncKitClient {
70 70 // ciphertext) arrives with the resumable multipart upload in a later pass.
71 71 drop(data);
72 72
73 - self.retry_request(Idempotency::Safe, || {
73 + self.retry_request(Idempotency::IdempotentWrite { on: "content hash (S3 key is content-addressed)" }, || {
74 74 let req = self
75 75 .http_stream
76 76 .put(presigned_url)
@@ -99,7 +99,7 @@ impl SyncKitClient {
99 99 size_bytes,
100 100 })?);
101 101
102 - self.retry_request(Idempotency::Safe, || {
102 + self.retry_request(Idempotency::IdempotentWrite { on: "blob hash" }, || {
103 103 let req = self
104 104 .http
105 105 .post(&self.endpoints.blobs_confirm)
@@ -122,7 +122,7 @@ impl SyncKitClient {
122 122 })?);
123 123
124 124 let download: BlobDownloadUrlResponse = self
125 - .retry_request_json(Idempotency::Safe, || {
125 + .retry_request_json(Idempotency::ReadOnly, || {
126 126 let req = self
127 127 .http
128 128 .post(&self.endpoints.blobs_download)
@@ -152,7 +152,7 @@ impl SyncKitClient {
152 152 #[instrument(skip(self, presigned_url))]
153 153 pub async fn blob_download(&self, expected_hash: &str, presigned_url: &str) -> Result<Vec<u8>> {
154 154 let resp = self
155 - .retry_request(Idempotency::Safe, || {
155 + .retry_request(Idempotency::ReadOnly, || {
156 156 let req = self.http_stream.get(presigned_url);
157 157 async move { check_response(req.send().await?).await }
158 158 })
M shared/synckit-client/src/client/encryption.rs +4 -4
@@ -19,14 +19,14 @@ impl SyncKitClient {
19 19 let (url, token) = self.key_url_and_token()?;
20 20
21 21 let result = self
22 - .retry_request(Idempotency::Safe, || {
22 + .retry_request(Idempotency::ReadOnly, || {
23 23 let req = self.http.get(url).bearer_auth(&token);
24 24 async move {
25 25 let resp = req.send().await?;
26 26 match resp.status().as_u16() {
27 27 200 | 404 => Ok(resp),
28 28 status => {
29 - let message = resp.text().await.unwrap_or_default();
29 + let message = crate::client::helpers::read_text_capped(resp, crate::client::helpers::MAX_CONTROL_BODY_BYTES).await;
30 30 Err(crate::error::SyncKitError::Server { status, message, retry_after_secs: None })
31 31 }
32 32 }
@@ -143,7 +143,7 @@ impl SyncKitClient {
143 143 expected_version,
144 144 })?);
145 145
146 - self.retry_request(Idempotency::Safe, || {
146 + self.retry_request(Idempotency::IdempotentWrite { on: "expected_version optimistic lock" }, || {
147 147 let req = self
148 148 .http
149 149 .put(url)
@@ -169,7 +169,7 @@ impl SyncKitClient {
169 169 pub(super) async fn get_server_key_full(&self) -> Result<GetKeyResponse> {
170 170 let (url, token) = self.key_url_and_token()?;
171 171
172 - self.retry_request_json(Idempotency::Safe, || {
172 + self.retry_request_json(Idempotency::ReadOnly, || {
173 173 let req = self.http.get(url).bearer_auth(&token);
174 174 async move { check_response(req.send().await?).await }
175 175 })
M shared/synckit-client/src/client/helpers.rs +105 -12
@@ -12,16 +12,31 @@ use super::{BASE_DELAY, MAX_RETRIES, SyncKitClient};
12 12 ///
13 13 /// Retry replays the request, so it is only safe when the server treats the
14 14 /// operation idempotently. Requiring this argument makes that judgement an
15 - /// explicit, reviewable decision: a new endpoint cannot be fed to a retry
16 - /// helper without naming why replay is harmless here — and the judgement now has
17 - /// runtime teeth, not just documentary value: [`may_retry`](Self::may_retry)
18 - /// gates the retry loop, so an [`Unsafe`](Self::Unsafe) operation physically
19 - /// cannot be auto-replayed.
15 + /// explicit, reviewable decision, and the judgement has runtime teeth:
16 + /// [`may_retry`](Self::may_retry) gates the retry loop, so an
17 + /// [`Unsafe`](Self::Unsafe) operation physically cannot be auto-replayed.
18 + ///
19 + /// The variants split "safe to replay" into a **read** and a **write that the
20 + /// server dedups**, and the write variant must *name* what makes replay
21 + /// idempotent. That is the forcing function: a mutating create tagged
22 + /// auto-retryable without naming its dedup key does not compile, so the
23 + /// `ota_create_release`/`ota_register_artifact` class — a `CREATE` that silently
24 + /// relied on a server unique constraint nobody had written down — cannot recur
25 + /// unnoticed.
20 26 #[derive(Clone, Copy, Debug)]
21 27 pub(super) enum Idempotency {
22 - /// Read-only, or a write whose replay yields the same end state (a `GET`,
23 - /// a `PUT`/`DELETE` of a fixed resource, or an upsert keyed on stable input).
24 - Safe,
28 + /// A read with no server-side effect (a `GET`, or a `POST` that only reads).
29 + /// Replay is trivially safe.
30 + ReadOnly,
31 + /// A write whose replay collapses to a single effect because the server
32 + /// dedups on `on` — a content-addressed key, an optimistic version, or a
33 + /// `UNIQUE` constraint. `on` documents *why* replay is harmless, at the call
34 + /// site, and naming it is mandatory.
35 + IdempotentWrite {
36 + /// The key/constraint the server dedups on (e.g. `"(app_id, version)
37 + /// unique"`). Documentation, surfaced in review and logs.
38 + on: &'static str,
39 + },
25 40 /// Carries a client-generated idempotency key (e.g. push's `batch_id`) so
26 41 /// the server collapses duplicate deliveries into one effect.
27 42 Keyed,
@@ -34,7 +49,81 @@ pub(super) enum Idempotency {
34 49 impl Idempotency {
35 50 /// Whether the retry helpers may replay this operation on a transient error.
36 51 fn may_retry(self) -> bool {
37 - matches!(self, Idempotency::Safe | Idempotency::Keyed)
52 + matches!(
53 + self,
54 + Idempotency::ReadOnly | Idempotency::IdempotentWrite { .. } | Idempotency::Keyed
55 + )
56 + }
57 +
58 + /// The documented dedup basis for an idempotent write (`None` for the other
59 + /// variants), surfaced in the retry log so a replayed request records *why*
60 + /// replay was safe.
61 + fn dedup_basis(self) -> Option<&'static str> {
62 + match self {
63 + Idempotency::IdempotentWrite { on } => Some(on),
64 + _ => None,
65 + }
66 + }
67 + }
68 +
69 + /// Upper bound on a control/JSON/error response body read into memory. Blob
70 + /// bodies stream through [`super::SyncKitClient::blob_download`]'s own 4 GiB cap;
71 + /// every *other* response — JSON control replies, OTA manifests, SSE-open errors,
72 + /// 4xx/5xx error bodies — is small, so a hostile or buggy server streaming a
73 + /// multi-gigabyte body into one of them is a pure OOM lever. 8 MiB is generous
74 + /// for any legitimate control body.
75 + pub(super) const MAX_CONTROL_BODY_BYTES: usize = 8 * 1024 * 1024;
76 +
77 + /// Read a response body into memory with a hard byte cap — the ONE sanctioned
78 + /// way to buffer a non-blob body.
79 + ///
80 + /// `reqwest`'s `Response::json`/`text`/`bytes` read the whole body with no size
81 + /// limit, so a hostile server can stream an arbitrarily large body and OOM the
82 + /// client (worst case: the public, unauthenticated OTA updater check). This
83 + /// drains `bytes_stream()` and aborts the instant the running total exceeds
84 + /// `limit`, and fast-rejects on an honest oversized `Content-Length` before
85 + /// reading a byte. Those raw reader methods are banned by `clippy.toml`
86 + /// (`disallowed-methods`) so a new call site cannot reintroduce an uncapped read.
87 + pub(super) async fn read_body_capped(resp: reqwest::Response, limit: usize) -> Result<bytes::Bytes> {
88 + use tokio_stream::StreamExt;
89 + if let Some(len) = resp.content_length()
90 + && len > limit as u64
91 + {
92 + return Err(SyncKitError::Internal(format!(
93 + "response Content-Length {len} exceeds {limit}-byte cap"
94 + )));
95 + }
96 + let mut stream = resp.bytes_stream();
97 + let mut buf = bytes::BytesMut::new();
98 + while let Some(chunk) = stream.next().await {
99 + let chunk = chunk.map_err(SyncKitError::Http)?;
100 + if buf.len() + chunk.len() > limit {
101 + return Err(SyncKitError::Internal(format!(
102 + "response body exceeds {limit}-byte cap"
103 + )));
104 + }
105 + buf.extend_from_slice(&chunk);
106 + }
107 + Ok(buf.freeze())
108 + }
109 +
110 + /// [`read_body_capped`] + JSON deserialize — the capped replacement for
111 + /// `resp.json::<T>()`.
112 + pub(super) async fn read_json_capped<T: serde::de::DeserializeOwned>(
113 + resp: reqwest::Response,
114 + limit: usize,
115 + ) -> Result<T> {
116 + let bytes = read_body_capped(resp, limit).await?;
117 + Ok(serde_json::from_slice(&bytes)?)
118 + }
119 +
120 + /// [`read_body_capped`] + lossy UTF-8 — the capped replacement for `resp.text()`
121 + /// on error bodies (which are only logged, so lossy decoding and a `""` fallback
122 + /// on an oversized/failed read are fine).
123 + pub(super) async fn read_text_capped(resp: reqwest::Response, limit: usize) -> String {
124 + match read_body_capped(resp, limit).await {
125 + Ok(bytes) => String::from_utf8_lossy(&bytes).into_owned(),
126 + Err(_) => String::new(),
38 127 }
39 128 }
40 129
@@ -100,6 +189,7 @@ impl SyncKitClient {
100 189 max_retries = MAX_RETRIES,
101 190 delay_ms = delay.as_millis() as u64,
102 191 error = %err,
192 + idempotency_basis = idempotency.dedup_basis(),
103 193 "Transient error, retrying after backoff",
104 194 );
105 195 tokio::time::sleep(delay).await;
@@ -136,10 +226,10 @@ impl SyncKitClient {
136 226 for attempt in 0..=max_attempts {
137 227 match operation().await {
138 228 Ok(resp) => {
139 - match resp.json::<T>().await {
229 + match read_json_capped::<T>(resp, MAX_CONTROL_BODY_BYTES).await {
140 230 Ok(parsed) => return Ok(parsed),
141 231 Err(e) => {
142 - let err = SyncKitError::Http(e);
232 + let err = e;
143 233 if attempt < max_attempts {
144 234 let delay = retry_delay(&err, attempt);
145 235 tracing::debug!(
@@ -400,7 +490,10 @@ pub(super) async fn check_response(resp: reqwest::Response) -> Result<reqwest::R
400 490 let status = resp.status().as_u16();
401 491 if status >= 400 {
402 492 let retry_after_secs = parse_retry_after(&resp);
403 - let message = resp.text().await.unwrap_or_default();
493 + // Capped read: this error path sits in front of every request (including
494 + // the otherwise-capped blob/SSE streams), so an uncapped `resp.text()`
495 + // here let a hostile server bypass those caps with a giant error body.
496 + let message = read_text_capped(resp, MAX_CONTROL_BODY_BYTES).await;
404 497 return Err(SyncKitError::Server { status, message, retry_after_secs });
405 498 }
406 499 Ok(resp)
M shared/synckit-client/src/client/mod.rs +90 -9
@@ -85,6 +85,13 @@ const BASE_DELAY: Duration = Duration::from_secs(1);
85 85 /// Avoids sending a request with a token that expires mid-flight.
86 86 const TOKEN_EXPIRY_BUFFER_SECS: i64 = 30;
87 87
88 + /// Inactivity (read) timeout for the streaming HTTP client. Bounds the gap
89 + /// between successive body reads without capping the total transfer, so a stalled
90 + /// (slow-loris) blob/OTA/SSE connection is torn down while a legitimately long,
91 + /// steady transfer is not. Kept above the server's 30s SSE keepalive interval so
92 + /// an idle notification stream is never mistaken for a stall.
93 + const STREAM_READ_TIMEOUT: Duration = Duration::from_secs(90);
94 +
88 95 /// Configuration for the SyncKit client.
89 96 #[derive(Debug, Clone)]
90 97 pub struct SyncKitConfig {
@@ -113,6 +120,12 @@ struct Endpoints {
113 120 subscription_storage_cap: String,
114 121 app_pricing: String,
115 122 account: String,
123 + /// Base for OTA paths (`{server}/api/v1/sync/ota`). The app-scoped and public
124 + /// OTA URLs carry runtime ids (app, release, slug/target/arch/version) so they
125 + /// cannot be pre-built like the static endpoints above; the `ota_*` builder
126 + /// methods construct them from this base, keeping *all* path construction in
127 + /// this one type instead of re-deriving it with `format!` in `ota.rs`.
128 + ota_base: String,
116 129 }
117 130
118 131 impl Endpoints {
@@ -136,8 +149,29 @@ impl Endpoints {
136 149 subscription_storage_cap: format!("{base}/api/v1/sync/subscription/storage-cap"),
137 150 app_pricing: format!("{base}/api/v1/sync/app/pricing"),
138 151 account: format!("{base}/api/v1/sync/account"),
152 + ota_base: format!("{base}/api/v1/sync/ota"),
139 153 }
140 154 }
155 +
156 + /// `POST` here to create a release, or list releases, for an app.
157 + fn ota_releases(&self, app_id: AppId) -> String {
158 + format!("{}/apps/{app_id}/releases", self.ota_base)
159 + }
160 +
161 + /// `POST` here to register an artifact under a release.
162 + fn ota_artifacts(&self, app_id: AppId, release_id: uuid::Uuid) -> String {
163 + format!("{}/apps/{app_id}/releases/{release_id}/artifacts", self.ota_base)
164 + }
165 +
166 + /// `POST` here to confirm an uploaded artifact under a release.
167 + fn ota_confirm(&self, app_id: AppId, release_id: uuid::Uuid) -> String {
168 + format!("{}/apps/{app_id}/releases/{release_id}/artifacts/confirm", self.ota_base)
169 + }
170 +
171 + /// The public, unauthenticated Tauri updater check URL.
172 + fn ota_updater(&self, slug: &str, target: &str, arch: &str, current_version: &str) -> String {
173 + format!("{}/{slug}/{target}/{arch}/{current_version}", self.ota_base)
174 + }
141 175 }
142 176
143 177 /// Session state obtained after authentication.
@@ -199,9 +233,16 @@ impl SyncKitClient {
199 233 .expect("failed to build HTTP client");
200 234
201 235 // No whole-request timeout: this client carries multi-gigabyte blobs and
202 - // the long-lived SSE stream, which a fixed 30s cap would break.
236 + // the long-lived SSE stream, which a fixed 30s cap would break. It DOES
237 + // carry a read (inactivity) timeout: without one, a server that dribbles
238 + // one byte an hour holds a blob/OTA/SSE connection open forever (slow-
239 + // loris). The cap bounds the gap *between* reads, not the total transfer,
240 + // so a legitimate slow-but-steady transfer is unaffected. 90s sits well
241 + // above the server's 30s SSE keepalive, so an idle notification stream
242 + // (which receives a keepalive comment every 30s) is never torn down.
203 243 let http_stream = Client::builder()
204 244 .connect_timeout(Duration::from_secs(10))
245 + .read_timeout(STREAM_READ_TIMEOUT)
205 246 .pool_max_idle_per_host(5)
206 247 .pool_idle_timeout(Duration::from_secs(90))
207 248 .https_only(https_only)
@@ -330,13 +371,36 @@ fn requires_https(server_url: &str) -> bool {
330 371 // https (or any non-plaintext scheme): enforcing https_only is a no-op.
331 372 return true;
332 373 };
333 - // Strictly loopback only — `0.0.0.0` is the unspecified/bind-all address, not
334 - // a loopback you'd connect *to*, so it does not belong in the plaintext-exempt
335 - // set.
336 - let is_loopback = rest.starts_with("localhost")
337 - || rest.starts_with("127.")
338 - || rest.starts_with("[::1]")
339 - || rest.starts_with("::1");
374 + // Extract the host *exactly*. A prefix match (`starts_with("127.")`,
375 + // `starts_with("localhost")`) trusts `http://127.0.0.1.attacker.com` and
376 + // `http://localhost.evil.com` — hostnames that resolve to an attacker's IP —
377 + // and would send the bearer token to them in cleartext. Parse the authority,
378 + // strip the port, and require an exact `localhost` or an IP literal that is
379 + // genuinely loopback. `0.0.0.0` is the unspecified/bind-all address, not a
380 + // loopback you connect *to*, so `is_loopback()` correctly rejects it.
381 + let authority = rest.split(['/', '?', '#']).next().unwrap_or(rest);
382 + // Strip userinfo: everything up to the last `@` is credentials, not the host.
383 + // `http://127.0.0.1:80@evil.com` connects to evil.com, so a naive port-split
384 + // on the raw authority would read `127.0.0.1` and wrongly exempt it (the same
385 + // userinfo bypass class as the server-side SSRF finding).
386 + let authority = authority.rsplit_once('@').map_or(authority, |(_, h)| h);
387 + let host = if let Some(after) = authority.strip_prefix('[') {
388 + // IPv6 literal `[::1]:port`: the host is inside the brackets, and the only
389 + // thing allowed after `]` is a port. Anything else (`[::1].evil.com`) is
390 + // malformed — fail closed.
391 + match after.split_once(']') {
392 + Some((h, rest)) if rest.is_empty() || rest.starts_with(':') => h,
393 + _ => return true,
394 + }
395 + } else {
396 + // `host:port` or bare `host` — cut at the port separator.
397 + authority.split(':').next().unwrap_or(authority)
398 + };
399 + let is_loopback = host == "localhost"
400 + || host
401 + .parse::<std::net::IpAddr>()
402 + .map(|ip| ip.is_loopback())
403 + .unwrap_or(false);
340 404 !is_loopback
341 405 }
342 406
@@ -369,7 +433,7 @@ pub async fn validate_api_key(server_url: &str, api_key: &str) -> Result<String>
369 433 struct ValidateResponse {
370 434 app_name: String,
371 435 }
372 - let body: ValidateResponse = resp.json().await?;
436 + let body: ValidateResponse = crate::client::helpers::read_json_capped(resp, crate::client::helpers::MAX_CONTROL_BODY_BYTES).await?;
373 437 Ok(body.app_name)
374 438 }
375 439
@@ -390,6 +454,23 @@ mod tests {
390 454 assert!(!requires_https("http://127.0.0.1:8080"));
391 455 assert!(!requires_https("http://localhost:3000"));
392 456 assert!(!requires_https("http://[::1]:9000"));
457 + assert!(!requires_https("http://127.0.0.1"));
458 + assert!(!requires_https("http://127.5.6.7:80/path")); // 127/8 is all loopback
459 + }
460 +
461 + #[test]
462 + fn requires_https_rejects_loopback_prefix_spoofs() {
463 + // The exact-host fix: a hostname that merely starts with a loopback
464 + // string but resolves elsewhere must NOT be exempted from TLS — the old
465 + // prefix match sent the bearer token to these in cleartext.
466 + assert!(requires_https("http://127.0.0.1.attacker.com"));
467 + assert!(requires_https("http://127.0.0.1.attacker.com:8080/pull"));
468 + assert!(requires_https("http://localhost.evil.com"));
469 + assert!(requires_https("http://localhostx"));
470 + assert!(requires_https("http://[::1].evil.com")); // junk after the bracket
471 + // userinfo bypass: the real host is evil.com, not the loopback userinfo.
472 + assert!(requires_https("http://127.0.0.1:80@evil.com"));
473 + assert!(requires_https("http://localhost@evil.com/pull"));
393 474 }
394 475
395 476 fn test_config() -> SyncKitConfig {
M shared/synckit-client/src/client/ota.rs +13 -21
@@ -70,13 +70,6 @@ struct RegisterArtifactBody<'a> {
70 70 }
71 71
72 72 impl SyncKitClient {
73 - /// Base URL for OTA management endpoints scoped to the authenticated app.
74 - fn ota_app_base(&self) -> Result<String> {
75 - let (app_id, _user_id) = self.require_session_ids()?;
76 - let base = self.config().server_url.trim_end_matches('/');
77 - Ok(format!("{base}/api/v1/sync/ota/apps/{app_id}"))
78 - }
79 -
80 73 /// Create a new OTA release for the authenticated app.
81 74 ///
82 75 /// Pass an empty `signature` only for unsigned platforms — Tauri's updater
@@ -90,7 +83,8 @@ impl SyncKitClient {
90 83 signature: &str,
91 84 ) -> Result<OtaRelease> {
92 85 let token = self.require_token()?;
93 - let url = format!("{}/releases", self.ota_app_base()?);
86 + let (app_id, _user_id) = self.require_session_ids()?;
87 + let url = self.endpoints.ota_releases(app_id);
94 88
95 89 let body = Bytes::from(serde_json::to_vec(&CreateReleaseBody {
96 90 version,
@@ -98,7 +92,7 @@ impl SyncKitClient {
98 92 signature,
99 93 })?);
100 94
101 - self.retry_request_json(Idempotency::Safe, || {
95 + self.retry_request_json(Idempotency::IdempotentWrite { on: "(app_id, version) unique — migration 033" }, || {
102 96 let req = self
103 97 .http
104 98 .post(&url)
@@ -123,7 +117,8 @@ impl SyncKitClient {
123 117 file_size: i64,
124 118 ) -> Result<OtaArtifactUpload> {
125 119 let token = self.require_token()?;
126 - let url = format!("{}/releases/{release_id}/artifacts", self.ota_app_base()?);
120 + let (app_id, _user_id) = self.require_session_ids()?;
121 + let url = self.endpoints.ota_artifacts(app_id, release_id);
127 122
128 123 let body = Bytes::from(serde_json::to_vec(&RegisterArtifactBody {
129 124 target,
@@ -131,7 +126,7 @@ impl SyncKitClient {
131 126 file_size,
132 127 })?);
133 128
134 - self.retry_request_json(Idempotency::Safe, || {
129 + self.retry_request_json(Idempotency::IdempotentWrite { on: "(release_id, target, arch) unique — migration 033" }, || {
135 130 let req = self
136 131 .http
137 132 .post(&url)
@@ -155,16 +150,14 @@ impl SyncKitClient {
155 150 arch: &str,
156 151 ) -> Result<()> {
157 152 let token = self.require_token()?;
158 - let url = format!(
159 - "{}/releases/{release_id}/artifacts/confirm",
160 - self.ota_app_base()?
161 - );
153 + let (app_id, _user_id) = self.require_session_ids()?;
154 + let url = self.endpoints.ota_confirm(app_id, release_id);
162 155 let body = Bytes::from(serde_json::to_vec(&serde_json::json!({
163 156 "target": target,
164 157 "arch": arch,
165 158 }))?);
166 159
167 - self.retry_request(Idempotency::Safe, || {
160 + self.retry_request(Idempotency::IdempotentWrite { on: "(release_id, target, arch)" }, || {
168 161 let req = self
169 162 .http
170 163 .post(&url)
@@ -183,7 +176,7 @@ impl SyncKitClient {
183 176 #[instrument(skip(self, presigned_url, data))]
184 177 pub async fn ota_upload_artifact(&self, presigned_url: &str, data: Vec<u8>) -> Result<()> {
185 178 let data = Bytes::from(data);
186 - self.retry_request(Idempotency::Safe, || {
179 + self.retry_request(Idempotency::IdempotentWrite { on: "S3 key (content-addressed)" }, || {
187 180 let req = self
188 181 .http_stream
189 182 .put(presigned_url)
@@ -208,11 +201,10 @@ impl SyncKitClient {
208 201 arch: &str,
209 202 current_version: &str,
210 203 ) -> Result<Option<OtaManifest>> {
211 - let base = self.config().server_url.trim_end_matches('/');
212 - let url = format!("{base}/api/v1/sync/ota/{slug}/{target}/{arch}/{current_version}");
204 + let url = self.endpoints.ota_updater(slug, target, arch, current_version);
213 205
214 206 let resp = self
215 - .retry_request(Idempotency::Safe, || {
207 + .retry_request(Idempotency::ReadOnly, || {
216 208 let req = self.http_stream.get(&url);
217 209 async move { check_response(req.send().await?).await }
218 210 })
@@ -222,7 +214,7 @@ impl SyncKitClient {
222 214 return Ok(None);
223 215 }
224 216
225 - Ok(Some(resp.json::<OtaManifest>().await?))
217 + Ok(Some(crate::client::helpers::read_json_capped::<OtaManifest>(resp, crate::client::helpers::MAX_CONTROL_BODY_BYTES).await?))
226 218 }
227 219 }
228 220
M shared/synckit-client/src/client/rotation.rs +48 -28
@@ -12,6 +12,15 @@ use crate::{
12 12 use super::SyncKitClient;
13 13 use super::helpers::{check_response, Idempotency};
14 14
15 + /// Hard ceiling on re-encrypt / straggler rounds within a single rotation. Each
16 + /// round re-encrypts one server batch and the set provably shrinks (re-encrypted
17 + /// entries drop out of the server's `key_id != new_key_id` filter), so a real
18 + /// rotation finishes in `total_entries / batch_size` rounds. The cap only fires
19 + /// when the set does NOT shrink — a stalled or hostile server that keeps handing
20 + /// back work — converting an unbounded spin into a bounded error. Generous enough
21 + /// to cover any real sync log at any batch size.
22 + const MAX_ROTATION_ROUNDS: u32 = 100_000;
23 +
15 24 impl SyncKitClient {
16 25 /// Rotate the master encryption key.
17 26 ///
@@ -67,36 +76,26 @@ impl SyncKitClient {
67 76 "Key rotation started",
68 77 );
69 78
70 - // 4. Re-encrypt loop
71 - loop {
72 - let all_done = self.reencrypt_batch(
73 - rotation_id,
74 - &old_key,
75 - &new_master_key,
76 - ).await?;
77 -
78 - if all_done {
79 - break;
80 - }
81 - }
79 + // 4. Re-encrypt loop (bounded — see reencrypt_until_done).
80 + self.reencrypt_until_done(rotation_id, &old_key, &new_master_key).await?;
82 81
83 - // 5. Complete — retry if stragglers from concurrent pushes
82 + // 5. Complete — retry if stragglers arrived from concurrent pushes. Both
83 + // the straggler retries and each re-encrypt pass are capped so a server
84 + // that keeps returning 409 (a stall, or a hostile peer racing pushes)
85 + // cannot spin this loop forever.
86 + let mut straggler_rounds = 0u32;
84 87 loop {
85 88 match self.complete_rotation(rotation_id).await {
86 89 Ok(()) => break,
87 90 Err(crate::error::SyncKitError::Server { status: 409, .. }) => {
88 - tracing::debug!("Stragglers detected, re-encrypting remaining entries");
89 - // Re-encrypt any new entries pushed during our rotation
90 - loop {
91 - let all_done = self.reencrypt_batch(
92 - rotation_id,
93 - &old_key,
94 - &new_master_key,
95 - ).await?;
96 - if all_done {
97 - break;
98 - }
91 + straggler_rounds += 1;
92 + if straggler_rounds >= MAX_ROTATION_ROUNDS {
93 + return Err(crate::error::SyncKitError::Internal(
94 + "key rotation did not converge: server kept reporting stragglers past the round cap".into(),
95 + ));
99 96 }
97 + tracing::debug!(round = straggler_rounds, "Stragglers detected, re-encrypting remaining entries");
98 + self.reencrypt_until_done(rotation_id, &old_key, &new_master_key).await?;
100 99 }
101 100 Err(e) => return Err(e),
102 101 }
@@ -135,6 +134,27 @@ impl SyncKitClient {
135 134 }
136 135 }
137 136
137 + /// Run [`reencrypt_batch`](Self::reencrypt_batch) until the server reports no
138 + /// entries left, bounded by [`MAX_ROTATION_ROUNDS`]. The batch set shrinks
139 + /// every round under an honest server; the cap is the backstop against a
140 + /// server that never reports it done, so the client fails with an error
141 + /// instead of spinning forever.
142 + async fn reencrypt_until_done(
143 + &self,
144 + rotation_id: Uuid,
145 + old_key: &[u8; 32],
146 + new_key: &[u8; 32],
147 + ) -> Result<()> {
148 + for _ in 0..MAX_ROTATION_ROUNDS {
149 + if self.reencrypt_batch(rotation_id, old_key, new_key).await? {
150 + return Ok(());
151 + }
152 + }
153 + Err(crate::error::SyncKitError::Internal(
154 + "key rotation did not converge: re-encrypt exceeded the round cap without the entry set draining".into(),
155 + ))
156 + }
157 +
138 158 /// Pull a batch of entries needing re-encryption, re-encrypt them, and push back.
139 159 /// Returns true if there are no more entries to process.
140 160 async fn reencrypt_batch(
@@ -204,7 +224,7 @@ impl SyncKitClient {
204 224 expected_key_version,
205 225 })?);
206 226
207 - self.retry_request_json(Idempotency::Safe, || {
227 + self.retry_request_json(Idempotency::IdempotentWrite { on: "one rotation per (app_id, user_id) — migration 089" }, || {
208 228 let req = self
209 229 .http
210 230 .post(&url)
@@ -229,7 +249,7 @@ impl SyncKitClient {
229 249 after_seq,
230 250 })?);
231 251
232 - self.retry_request_json(Idempotency::Safe, || {
252 + self.retry_request_json(Idempotency::ReadOnly, || {
233 253 let req = self
234 254 .http
235 255 .post(&url)
@@ -254,7 +274,7 @@ impl SyncKitClient {
254 274 entries,
255 275 })?);
256 276
257 - self.retry_request_json(Idempotency::Safe, || {
277 + self.retry_request_json(Idempotency::IdempotentWrite { on: "seq (re-encrypt overwrites the same row)" }, || {
258 278 let req = self
259 279 .http
260 280 .post(&url)
@@ -277,7 +297,7 @@ impl SyncKitClient {
277 297 rotation_id,
278 298 })?);
279 299
280 - self.retry_request(Idempotency::Safe, || {
300 + self.retry_request(Idempotency::IdempotentWrite { on: "rotation_id" }, || {
281 301 let req = self
282 302 .http
283 303 .post(&url)
M shared/synckit-client/src/client/subscribe.rs +19 -2
@@ -43,6 +43,14 @@ const MAX_SSE_BUFFER: usize = 1024 * 1024;
43 43 /// Cap on the reconnect backoff delay.
44 44 const MAX_RECONNECT_DELAY: Duration = Duration::from_secs(60);
45 45
46 + /// Maximum *consecutive* failed reconnect attempts before the stream gives up.
47 + /// A single success resets the counter, so this only fires when the server has
48 + /// been unreachable or erroring for a sustained stretch (with the 60s backoff
49 + /// cap, several minutes). Without it, a permanently-degraded non-auth server
50 + /// (persistent 500s, connection refused) is retried silently forever — the
51 + /// caller is better told the stream is dead so it can re-subscribe later.
52 + const MAX_RECONNECT_ATTEMPTS: u32 = 10;
53 +
46 54 /// Reconnect backoff: 1s, 2s, 4s, ... capped at [`MAX_RECONNECT_DELAY`], with
47 55 /// +/-20% jitter so a fleet of clients doesn't reconnect in lockstep.
48 56 fn reconnect_delay(attempt: u32) -> Duration {
@@ -74,10 +82,19 @@ impl SyncNotifyStream {
74 82
75 83 /// Re-establish the SSE connection after a drop. Loops with backoff on
76 84 /// transient failures; returns `false` (fatal) if the server rejects the
77 - /// request for auth reasons, so the caller stops and re-authenticates.
85 + /// request for auth reasons (caller re-authenticates) or after
86 + /// [`MAX_RECONNECT_ATTEMPTS`] consecutive failures (caller re-subscribes
87 + /// later) rather than retrying a dead server forever.
78 88 async fn reconnect(&mut self) -> bool {
79 89 loop {
80 90 self.reconnect_attempt += 1;
91 + if self.reconnect_attempt > MAX_RECONNECT_ATTEMPTS {
92 + tracing::warn!(
93 + attempts = self.reconnect_attempt - 1,
94 + "SSE reconnect gave up after too many consecutive failures; stream closing"
95 + );
96 + return false;
97 + }
81 98 tokio::time::sleep(reconnect_delay(self.reconnect_attempt)).await;
82 99
83 100 match self.http.get(&self.url).bearer_auth(&*self.token).send().await {
@@ -217,7 +234,7 @@ impl SyncKitClient {
217 234
218 235 let status = resp.status().as_u16();
219 236 if status >= 400 {
220 - let message = resp.text().await.unwrap_or_default();
237 + let message = crate::client::helpers::read_text_capped(resp, crate::client::helpers::MAX_CONTROL_BODY_BYTES).await;
221 238 return Err(SyncKitError::Server { status, message, retry_after_secs: None });
222 239 }
223 240
M shared/synckit-client/src/client/subscription.rs +5 -5
@@ -234,7 +234,7 @@ impl SyncKitClient {
234 234 let body = Bytes::from(serde_json::to_vec(&AppPricingRequest {
235 235 api_key: &self.config.api_key,
236 236 })?);
237 - self.retry_request_json(Idempotency::Safe, || {
237 + self.retry_request_json(Idempotency::ReadOnly, || {
238 238 let req = self
239 239 .http
240 240 .post(&self.endpoints.app_pricing)
@@ -255,7 +255,7 @@ impl SyncKitClient {
255 255 cap_bytes,
256 256 interval: interval.as_str(),
257 257 })?);
258 - self.retry_request_json(Idempotency::Safe, || {
258 + self.retry_request_json(Idempotency::ReadOnly, || {
259 259 let req = self
260 260 .http
261 261 .post(&self.endpoints.subscription_quote)
@@ -273,7 +273,7 @@ impl SyncKitClient {
273 273 #[tracing::instrument(skip(self))]
274 274 pub async fn get_account_info(&self) -> Result<AccountInfo> {
275 275 let token = self.require_token()?;
276 - self.retry_request_json(Idempotency::Safe, || {
276 + self.retry_request_json(Idempotency::ReadOnly, || {
277 277 let req = self.http.get(&self.endpoints.account).bearer_auth(token.as_str());
278 278 async move { check_response(req.send().await?).await }
279 279 })
@@ -287,7 +287,7 @@ impl SyncKitClient {
287 287 #[tracing::instrument(skip(self))]
288 288 pub async fn get_subscription_status(&self) -> Result<SubscriptionStatus> {
289 289 let token = self.require_token()?;
290 - self.retry_request_json(Idempotency::Safe, || {
290 + self.retry_request_json(Idempotency::ReadOnly, || {
291 291 let req = self.http.get(&self.endpoints.subscription).bearer_auth(token.as_str());
292 292 async move { check_response(req.send().await?).await }
293 293 })
@@ -329,7 +329,7 @@ impl SyncKitClient {
329 329 pub async fn queue_storage_cap_change(&self, cap_bytes: i64) -> Result<SubscriptionStatus> {
330 330 let token = self.require_token()?;
331 331 let body = Bytes::from(serde_json::to_vec(&CapChangeRequest { cap_bytes })?);
332 - self.retry_request_json(Idempotency::Safe, || {
332 + self.retry_request_json(Idempotency::IdempotentWrite { on: "pending cap change (upsert)" }, || {
333 333 let req = self
334 334 .http
335 335 .post(&self.endpoints.subscription_storage_cap)
M shared/synckit-client/src/client/sync.rs +4 -4
@@ -33,7 +33,7 @@ impl SyncKitClient {
33 33 platform: platform.to_string(),
34 34 })?);
35 35
36 - self.retry_request_json(Idempotency::Safe, || {
36 + self.retry_request_json(Idempotency::IdempotentWrite { on: "device registration (server-deduped)" }, || {
37 37 let req = self
38 38 .http
39 39 .post(&self.endpoints.devices)
@@ -50,7 +50,7 @@ impl SyncKitClient {
50 50 pub async fn list_devices(&self) -> Result<Vec<Device>> {
51 51 let token = self.require_token()?;
52 52
53 - self.retry_request_json(Idempotency::Safe, || {
53 + self.retry_request_json(Idempotency::ReadOnly, || {
54 54 let req = self.http.get(&self.endpoints.devices).bearer_auth(&token);
55 55 async move { check_response(req.send().await?).await }
56 56 })
@@ -198,7 +198,7 @@ impl SyncKitClient {
198 198 let token = self.require_token()?;
199 199
200 200 let pull_resp: PullResponse = self
201 - .retry_request_json(Idempotency::Safe, || {
201 + .retry_request_json(Idempotency::ReadOnly, || {
202 202 let req = self
203 203 .http
204 204 .post(&self.endpoints.pull)
@@ -256,7 +256,7 @@ impl SyncKitClient {
256 256 pub async fn status(&self) -> Result<SyncStatus> {
257 257 let token = self.require_token()?;
258 258
259 - self.retry_request_json(Idempotency::Safe, || {
259 + self.retry_request_json(Idempotency::ReadOnly, || {
260 260 let req = self.http.get(&self.endpoints.status).bearer_auth(&token);
261 261 async move { check_response(req.send().await?).await }
262 262 })
M shared/synckit-client/tests/integration.rs +23
@@ -82,6 +82,29 @@ fn device_json() -> serde_json::Value {
82 82 })
83 83 }
84 84
85 + // ── Response body cap (DoS) ──
86 +
87 + #[tokio::test]
88 + async fn oversized_response_body_is_capped_not_buffered() {
89 + // A hostile/buggy server streams a control-plane body far larger than the
90 + // 8 MiB cap. The client must reject it (the cap fast-rejects on the honest
91 + // Content-Length) instead of buffering it into memory and OOMing.
92 + let server = MockServer::start().await;
93 + let huge = vec![b'x'; 9 * 1024 * 1024];
94 + Mock::given(method("GET"))
95 + .and(path("/api/v1/sync/status"))
96 + .respond_with(ResponseTemplate::new(200).set_body_bytes(huge))
97 + .mount(&server)
98 + .await;
99 +
100 + let client = authed_client(&server);
101 + let err = client.status().await.unwrap_err();
102 + assert!(
103 + matches!(err, SyncKitError::Internal(ref m) if m.contains("cap")),
104 + "expected the body cap to reject the oversized response, got: {err:?}"
105 + );
106 + }
107 +
85 108 // ── Auth flow ──
86 109
87 110 #[tokio::test]