Skip to main content

max / makenotwork

synckit: enforce paid-only first-party sync + authoritative blob size + atomic caps (ultra-fuzz Run 3 #1/#2/#3) Phase 1 of the SyncKit ultra-fuzz remediation (Payments + Storage axes). #1 CRITICAL — first-party (is_internal) apps used the end-user subscription model but enforced nothing: any logged-in user could mint a token and upload unlimited blobs free. Add db::synckit::internal_write_allowed and gate the write paths (blob presign, blob confirm, push) on an active app_sync_subscriptions row. confirm_internal_blob enforces the per-user storage_limit_bytes atomically (lock the sub row, sum authoritative sync_blobs usage, insert) — paid-only sync. Reads/pulls stay open so a lapsed user can still export their data. #2 SERIOUS — blob confirm trusted the client-declared size_bytes and the presign left Content-Length unbound. Bind max_bytes at presign and read the authoritative object_size at confirm; size_bytes is dropped from BlobConfirmRequest. #3 SERIOUS — TOCTOU over-allocation. key_cap is now checked inside claim_key under the usage-row FOR UPDATE lock (cap_reached outcome), not pre-transaction. confirm_developer_blob folds would_exceed_storage + add_bytes_stored into one locked transaction so concurrent uploads can't overshoot the cap. Band-aids removed: would_exceed_storage, add_bytes_stored, ExceededLimit, is_key_actively_claimed, create_sync_blob_idempotent. Also implement the missing download_object_buf on the test storage mock. Tests: new synckit_paid_sync.rs (no-sub gate on push/upload/confirm, cap enforcement, size-spoof caught by object_size); per_key/adversarial helpers updated to store authoritative bytes. 49/49 synckit tests pass. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-23 16:05 UTC
Commit: 5a9a45293a031e6a695773d00b777e70f15faaa6
Parent: d49b61c
12 files changed, +648 insertions, -321 deletions
M server/src/db/synckit/blobs.rs +212 -23
@@ -26,15 +26,31 @@ pub async fn get_sync_blob_by_hash(
26 26 Ok(blob)
27 27 }
28 28
29 - /// Insert a sync blob, updating size on conflict (idempotent).
30 - ///
31 - /// Uses `ON CONFLICT DO UPDATE` to keep `size_bytes` consistent with the
32 - /// actual S3 object when concurrent uploads race. `key` is the developer-
33 - /// defined SDK key from the session JWT; it never changes on re-upload of
34 - /// the same hash so the conflict path leaves it alone.
35 - #[tracing::instrument(skip_all)]
36 - pub async fn create_sync_blob_idempotent(
37 - pool: &PgPool,
29 + /// Result of an atomic blob-confirm. Every variant is terminal for one confirm
30 + /// call; the route handler maps it to a 204 or a 402 with the right reason.
31 + #[derive(Debug, Clone, PartialEq, Eq)]
32 + pub enum BlobConfirm {
33 + /// Newly recorded; usage counters were incremented.
34 + Stored,
35 + /// The blob (same hash) was already recorded — idempotent re-confirm, no
36 + /// double counting.
37 + AlreadyStored,
38 + /// First-party app and the user has no `active` subscription (paid-only).
39 + NoSubscription,
40 + /// Storing this blob would exceed the applicable cap. Nothing was written.
41 + QuotaExceeded {
42 + dimension: &'static str,
43 + used: i64,
44 + limit: i64,
45 + key: Option<String>,
46 + },
47 + }
48 +
49 + /// Insert a blob row inside an open transaction. `ON CONFLICT DO UPDATE` keeps
50 + /// `size_bytes` consistent with the actual S3 object; the idempotency check in
51 + /// the callers means we only reach this for a genuinely new `(app,user,hash)`.
52 + async fn insert_blob_tx(
53 + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
38 54 app_id: SyncAppId,
39 55 user_id: UserId,
40 56 hash: &str,
@@ -56,33 +72,206 @@ pub async fn create_sync_blob_idempotent(
56 72 .bind(size_bytes)
57 73 .bind(s3_key)
58 74 .bind(key)
59 - .execute(pool)
75 + .execute(&mut **tx)
60 76 .await?;
61 -
62 77 Ok(())
63 78 }
64 79
65 - /// Get total blob storage used by a user for an app (in bytes).
66 - ///
67 - /// Currently unused at the call-site level (Phase 5 moved cap enforcement to
68 - /// the app-level counters in `sync_app_usage_current`), but kept around for
69 - /// the eventual per-user "what's taking up space" dashboard view.
70 - #[allow(dead_code)]
80 + /// Confirm a blob for a first-party (`is_internal`) app under the paid-only
81 + /// end-user model. Atomic: locks the user's subscription row, enforces an
82 + /// `active` status and the per-user `storage_limit_bytes`, then inserts — all
83 + /// in one transaction, so concurrent confirms for the same user can't overshoot
84 + /// the cap (the prior read-then-add gate could). Usage is summed from the
85 + /// authoritative `sync_blobs` table, so there is no counter to drift.
71 86 #[tracing::instrument(skip_all)]
72 - pub async fn get_blob_storage_used(
87 + pub async fn confirm_internal_blob(
73 88 pool: &PgPool,
74 89 app_id: SyncAppId,
75 90 user_id: UserId,
76 - ) -> Result<i64> {
77 - let total: Option<i64> = sqlx::query_scalar(
78 - "SELECT SUM(size_bytes)::BIGINT FROM sync_blobs WHERE app_id = $1 AND user_id = $2",
91 + hash: &str,
92 + size_bytes: i64,
93 + s3_key: &str,
94 + key: &str,
95 + ) -> Result<BlobConfirm> {
96 + let mut tx = pool.begin().await?;
97 +
98 + // Lock the subscription row for this user+app. Serializes this user's
99 + // confirms; different users don't contend.
100 + let sub: Option<(String, Option<i64>)> = sqlx::query_as(
101 + "SELECT status, storage_limit_bytes FROM app_sync_subscriptions
102 + WHERE app_id = $1 AND user_id = $2 FOR UPDATE",
79 103 )
80 104 .bind(app_id)
81 105 .bind(user_id)
82 - .fetch_one(pool)
106 + .fetch_optional(&mut *tx)
107 + .await?;
108 + let Some((status, limit)) = sub else {
109 + return Ok(BlobConfirm::NoSubscription);
110 + };
111 + if status != "active" {
112 + return Ok(BlobConfirm::NoSubscription);
113 + }
114 +
115 + // Idempotent re-confirm: already recorded, don't recount.
116 + let exists: Option<i32> = sqlx::query_scalar(
117 + "SELECT 1 FROM sync_blobs WHERE app_id = $1 AND user_id = $2 AND hash = $3",
118 + )
119 + .bind(app_id)
120 + .bind(user_id)
121 + .bind(hash)
122 + .fetch_optional(&mut *tx)
123 + .await?;
124 + if exists.is_some() {
125 + tx.commit().await?;
126 + return Ok(BlobConfirm::AlreadyStored);
127 + }
128 +
129 + let used: i64 = sqlx::query_scalar(
130 + "SELECT COALESCE(SUM(size_bytes), 0)::BIGINT FROM sync_blobs
131 + WHERE app_id = $1 AND user_id = $2",
132 + )
133 + .bind(app_id)
134 + .bind(user_id)
135 + .fetch_one(&mut *tx)
136 + .await?;
137 + let limit = limit.unwrap_or(0);
138 + if used.saturating_add(size_bytes) > limit {
139 + return Ok(BlobConfirm::QuotaExceeded {
140 + dimension: "storage",
141 + used,
142 + limit,
143 + key: None,
144 + });
145 + }
146 +
147 + insert_blob_tx(&mut tx, app_id, user_id, hash, size_bytes, s3_key, key).await?;
148 + tx.commit().await?;
149 + Ok(BlobConfirm::Stored)
150 + }
151 +
152 + /// Confirm a blob for a developer-billed (non-internal) app. Atomic: locks the
153 + /// app usage row (and the per-key row in `per_key` mode), re-checks the cap
154 + /// under the lock, inserts, and increments the counters — folding the old
155 + /// `would_exceed_storage` (read) + `add_bytes_stored` (add) pair into one
156 + /// transaction so concurrent uploads can't slip past the cap. Re-confirms are
157 + /// idempotent and never double-count.
158 + #[tracing::instrument(skip_all)]
159 + #[allow(clippy::too_many_arguments)]
160 + pub async fn confirm_developer_blob(
161 + pool: &PgPool,
162 + app_id: SyncAppId,
163 + user_id: UserId,
164 + hash: &str,
165 + size_bytes: i64,
166 + s3_key: &str,
167 + key: &str,
168 + enforcement_mode: &str,
169 + storage_gb_cap: Option<i32>,
170 + key_cap: Option<i32>,
171 + gb_per_key: Option<i32>,
172 + ) -> Result<BlobConfirm> {
173 + let mut tx = pool.begin().await?;
174 +
175 + // Lock the app usage row; this is the serialization point for the app-wide
176 + // counter. Returns current app-level bytes_stored.
177 + let app_used: i64 = sqlx::query_scalar(
178 + "SELECT bytes_stored FROM sync_app_usage_current WHERE app_id = $1 FOR UPDATE",
179 + )
180 + .bind(app_id)
181 + .fetch_one(&mut *tx)
182 + .await?;
183 +
184 + // Idempotent re-confirm.
185 + let exists: Option<i32> = sqlx::query_scalar(
186 + "SELECT 1 FROM sync_blobs WHERE app_id = $1 AND user_id = $2 AND hash = $3",
187 + )
188 + .bind(app_id)
189 + .bind(user_id)
190 + .bind(hash)
191 + .fetch_optional(&mut *tx)
192 + .await?;
193 + if exists.is_some() {
194 + tx.commit().await?;
195 + return Ok(BlobConfirm::AlreadyStored);
196 + }
197 +
198 + match enforcement_mode {
199 + "bulk" => {
200 + if let Some(gb) = storage_gb_cap {
201 + let limit = crate::synckit_billing::storage_cap_bytes(gb as u32);
202 + if app_used.saturating_add(size_bytes) > limit {
203 + return Ok(BlobConfirm::QuotaExceeded {
204 + dimension: "storage",
205 + used: app_used,
206 + limit,
207 + key: None,
208 + });
209 + }
210 + }
211 + }
212 + "per_key" => {
213 + if let (Some(kc), Some(g)) = (key_cap, gb_per_key) {
214 + let per_key_limit = crate::synckit_billing::storage_cap_bytes(g as u32);
215 + let app_limit =
216 + crate::synckit_billing::storage_cap_bytes(kc.saturating_mul(g) as u32);
217 + let key_used: i64 = sqlx::query_scalar(
218 + "SELECT bytes_stored FROM sync_key_usage_current
219 + WHERE app_id = $1 AND key = $2 FOR UPDATE",
220 + )
221 + .bind(app_id)
222 + .bind(key)
223 + .fetch_optional(&mut *tx)
224 + .await?
225 + .unwrap_or(0);
226 + if key_used.saturating_add(size_bytes) > per_key_limit {
227 + return Ok(BlobConfirm::QuotaExceeded {
228 + dimension: "storage_per_key",
229 + used: key_used,
230 + limit: per_key_limit,
231 + key: Some(key.to_string()),
232 + });
233 + }
234 + // Defensive app-aggregate ceiling (guards counter drift).
235 + if app_used.saturating_add(size_bytes) > app_limit {
236 + return Ok(BlobConfirm::QuotaExceeded {
237 + dimension: "storage",
238 + used: app_used,
239 + limit: app_limit,
240 + key: None,
241 + });
242 + }
243 + }
244 + }
245 + _ => {}
246 + }
247 +
248 + insert_blob_tx(&mut tx, app_id, user_id, hash, size_bytes, s3_key, key).await?;
249 +
250 + // Increment the app-wide and per-key counters in the same transaction.
251 + sqlx::query(
252 + "UPDATE sync_app_usage_current
253 + SET bytes_stored = GREATEST(bytes_stored + $2, 0), updated_at = NOW()
254 + WHERE app_id = $1",
255 + )
256 + .bind(app_id)
257 + .bind(size_bytes)
258 + .execute(&mut *tx)
259 + .await?;
260 + sqlx::query(
261 + "INSERT INTO sync_key_usage_current (app_id, key, bytes_stored)
262 + VALUES ($1, $2, GREATEST($3, 0))
263 + ON CONFLICT (app_id, key)
264 + DO UPDATE SET bytes_stored = GREATEST(sync_key_usage_current.bytes_stored + $3, 0),
265 + updated_at = NOW()",
266 + )
267 + .bind(app_id)
268 + .bind(key)
269 + .bind(size_bytes)
270 + .execute(&mut *tx)
83 271 .await?;
84 272
85 - Ok(total.unwrap_or(0))
273 + tx.commit().await?;
274 + Ok(BlobConfirm::Stored)
86 275 }
87 276
88 277 /// Count devices registered for a user/app pair.
M server/src/db/synckit/subscriptions.rs +35
@@ -28,6 +28,41 @@ pub struct DbAppSyncSubscription {
28 28 pub current_period_end: Option<DateTime<Utc>>,
29 29 }
30 30
31 + /// Whether writes (push / blob upload) are allowed for this `(app, user)`.
32 + ///
33 + /// Non-internal apps are developer-billed: the developer pays for the whole
34 + /// app's storage budget, so writes are always allowed here (storage caps are
35 + /// enforced separately on the blob path). First-party (`is_internal`) apps use
36 + /// the end-user subscription model — paid-only sync — so a write is allowed
37 + /// only when the user holds an `active` row in `app_sync_subscriptions`.
38 + ///
39 + /// A missing app resolves to `false` (deny); `app_id` always comes from a
40 + /// validated JWT at the call sites, so this is a defensive default.
41 + #[tracing::instrument(skip_all)]
42 + pub async fn internal_write_allowed(
43 + pool: &PgPool,
44 + app_id: SyncAppId,
45 + user_id: UserId,
46 + ) -> Result<bool> {
47 + let row: Option<(bool, bool)> = sqlx::query_as(
48 + r#"
49 + SELECT
50 + sa.is_internal,
51 + EXISTS(
52 + SELECT 1 FROM app_sync_subscriptions s
53 + WHERE s.app_id = sa.id AND s.user_id = $2 AND s.status = 'active'
54 + ) AS has_active_sub
55 + FROM sync_apps sa
56 + WHERE sa.id = $1
57 + "#,
58 + )
59 + .bind(app_id)
60 + .bind(user_id)
61 + .fetch_optional(pool)
62 + .await?;
63 + Ok(matches!(row, Some((is_internal, has_sub)) if !is_internal || has_sub))
64 + }
65 +
31 66 /// Look up the active subscription a user has on an app, if any.
32 67 /// Returns `None` if the user has never subscribed or the subscription was deleted.
33 68 #[tracing::instrument(skip_all)]
M server/src/db/synckit_billing.rs +48 -210
@@ -36,8 +36,13 @@ use crate::error::Result;
36 36 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
37 37 pub struct ClaimResult {
38 38 /// `true` if this call inserted a new active claim row;
39 - /// `false` if the key was already actively claimed (idempotent re-claim).
39 + /// `false` if the key was already actively claimed (idempotent re-claim)
40 + /// or the cap was reached (see `cap_reached`).
40 41 pub newly_claimed: bool,
42 + /// `true` if a new slot was refused because `key_cap` is already reached.
43 + /// Mutually exclusive with `newly_claimed`. An idempotent re-claim of an
44 + /// already-active key never sets this (it consumes no new slot).
45 + pub cap_reached: bool,
41 46 /// Total active claims for this app after the operation.
42 47 pub total_claimed: i32,
43 48 }
@@ -370,15 +375,19 @@ pub async fn get_apps_with_billing_by_project(
370 375 /// Claim an SDK encryption key for an app. Idempotent: a re-claim of an
371 376 /// already-active key returns `newly_claimed = false` without inserting.
372 377 ///
373 - /// The transaction locks `sync_app_usage_current` for this app first, so the
374 - /// route handler can read `keys_claimed` and decide on the cap before this
375 - /// runs without races. (The handler does its check pre-transaction; this
376 - /// function re-checks idempotency inside the lock.)
378 + /// `key_cap` is the per-app active-key ceiling (`Some` only for `per_key`
379 + /// developer apps; `None` means uncapped). The cap is checked **inside** the
380 + /// transaction, under the `FOR UPDATE` lock on `sync_app_usage_current`, so
381 + /// concurrent claims of distinct new keys can't collectively overshoot the cap
382 + /// — the prior design checked the cap pre-transaction in the handler and could
383 + /// over-allocate. A re-claim of an already-active key is admitted regardless of
384 + /// the cap (it consumes no new slot).
377 385 #[tracing::instrument(skip_all)]
378 386 pub async fn claim_key(
379 387 pool: &sqlx::PgPool,
380 388 app_id: SyncAppId,
381 389 key: &str,
390 + key_cap: Option<i32>,
382 391 ) -> Result<ClaimResult> {
383 392 let mut tx = pool.begin().await?;
384 393
@@ -400,33 +409,47 @@ pub async fn claim_key(
400 409 .fetch_optional(&mut *tx)
401 410 .await?;
402 411
403 - let newly_claimed = if existing.is_some() {
404 - false
405 - } else {
406 - sqlx::query(
407 - "INSERT INTO sync_app_keys (app_id, key) VALUES ($1, $2)",
408 - )
409 - .bind(app_id)
410 - .bind(key)
411 - .execute(&mut *tx)
412 - .await?;
412 + if existing.is_some() {
413 + // Idempotent re-claim — no new slot, cap not consulted.
414 + tx.commit().await?;
415 + return Ok(ClaimResult {
416 + newly_claimed: false,
417 + cap_reached: false,
418 + total_claimed: keys_claimed,
419 + });
420 + }
413 421
414 - sqlx::query(
415 - "UPDATE sync_app_usage_current
416 - SET keys_claimed = keys_claimed + 1, updated_at = NOW()
417 - WHERE app_id = $1",
418 - )
422 + // New claim: enforce the cap under the lock.
423 + if let Some(cap) = key_cap
424 + && keys_claimed >= cap
425 + {
426 + tx.commit().await?;
427 + return Ok(ClaimResult {
428 + newly_claimed: false,
429 + cap_reached: true,
430 + total_claimed: keys_claimed,
431 + });
432 + }
433 +
434 + sqlx::query("INSERT INTO sync_app_keys (app_id, key) VALUES ($1, $2)")
419 435 .bind(app_id)
436 + .bind(key)
420 437 .execute(&mut *tx)
421 438 .await?;
422 -
423 - keys_claimed += 1;
424 - true
425 - };
439 + sqlx::query(
440 + "UPDATE sync_app_usage_current
441 + SET keys_claimed = keys_claimed + 1, updated_at = NOW()
442 + WHERE app_id = $1",
443 + )
444 + .bind(app_id)
445 + .execute(&mut *tx)
446 + .await?;
447 + keys_claimed += 1;
426 448
427 449 tx.commit().await?;
428 450 Ok(ClaimResult {
429 - newly_claimed,
451 + newly_claimed: true,
452 + cap_reached: false,
430 453 total_claimed: keys_claimed,
431 454 })
432 455 }
@@ -547,23 +570,6 @@ pub async fn get_top_keys_per_app(
547 570
548 571 // ── Phase 5: Usage counters and cap enforcement ──
549 572
550 - /// A breached cap, returned by `would_exceed_*` checks.
551 - #[derive(Debug, Clone, PartialEq, Eq)]
552 - pub struct ExceededLimit {
553 - /// `"storage"`, `"storage_per_key"`, or `"egress"`. (`"billing"` is reserved
554 - /// for inactive-billing failure paths so a single 402 response shape can
555 - /// carry every reason; the caller decides whether to thread that through
556 - /// here or check `billing_status` separately.)
557 - pub dimension: &'static str,
558 - /// Current usage in bytes (before the would-be addition).
559 - pub used: i64,
560 - /// Configured cap in bytes.
561 - pub limit: i64,
562 - /// The SDK key that hit its cap, if `dimension == "storage_per_key"`.
563 - /// `None` for app-wide dimensions.
564 - pub key: Option<String>,
565 - }
566 -
567 573 /// A row that may need a warning email sent.
568 574 ///
569 575 /// `threshold_pct` is the highest WARNING_THRESHOLDS_PCT band currently
@@ -585,55 +591,6 @@ pub struct WarningCandidate {
585 591 pub key: Option<String>,
586 592 }
587 593
588 - /// Increment (or decrement, for negative `delta`) the rolling `bytes_stored`
589 - /// counter on `sync_app_usage_current` and the per-key row on
590 - /// `sync_key_usage_current` (upserted if missing). Returns the new app-level
591 - /// total.
592 - ///
593 - /// NOTE: read-then-add elsewhere in the request handler is racy. Acceptable
594 - /// overshoot under concurrent uploads in v1.
595 - #[tracing::instrument(skip_all)]
596 - pub async fn add_bytes_stored(
597 - pool: &PgPool,
598 - app_id: SyncAppId,
599 - key: &str,
600 - delta: i64,
601 - ) -> Result<i64> {
602 - let mut tx = pool.begin().await?;
603 -
604 - let (new_val,): (i64,) = sqlx::query_as(
605 - r#"
606 - UPDATE sync_app_usage_current
607 - SET bytes_stored = GREATEST(bytes_stored + $2, 0), updated_at = NOW()
608 - WHERE app_id = $1
609 - RETURNING bytes_stored
610 - "#,
611 - )
612 - .bind(app_id)
613 - .bind(delta)
614 - .fetch_one(&mut *tx)
615 - .await?;
616 -
617 - sqlx::query(
618 - r#"
619 - INSERT INTO sync_key_usage_current (app_id, key, bytes_stored)
620 - VALUES ($1, $2, GREATEST($3, 0))
621 - ON CONFLICT (app_id, key)
622 - DO UPDATE SET
623 - bytes_stored = GREATEST(sync_key_usage_current.bytes_stored + $3, 0),
624 - updated_at = NOW()
625 - "#,
626 - )
627 - .bind(app_id)
628 - .bind(key)
629 - .bind(delta)
630 - .execute(&mut *tx)
631 - .await?;
632 -
633 - tx.commit().await?;
634 - Ok(new_val)
635 - }
636 -
637 594 /// Increment (or decrement) the per-period `bytes_egress_period` counter.
638 595 /// Returns the new value.
639 596 #[tracing::instrument(skip_all)]
@@ -657,107 +614,6 @@ pub async fn add_bytes_egress(
657 614 Ok(new_val)
658 615 }
659 616
660 - /// Check whether storing `additional_bytes` more would exceed the storage cap.
661 - ///
662 - /// Returns `None` if no cap applies (internal apps, billing not active, or
663 - /// the cap is unset). Returns `Some(ExceededLimit)` only when adding
664 - /// `additional_bytes` would push usage past the cap.
665 - ///
666 - /// In `per_key` mode the per-key counter is checked first against
667 - /// `gb_per_key × 1GB` — if exceeded, returns `dimension: "storage_per_key"`
668 - /// with the offending `key`. The app-aggregate (`key_cap × gb_per_key`) is
669 - /// also checked as a defensive hard ceiling; under normal operation this
670 - /// can't trip before some key has tripped, but it guards against drift.
671 - ///
672 - /// In `bulk` mode `key` is unused (still passed in for call-site uniformity)
673 - /// and the app-wide `storage_gb_cap` is the only check.
674 - ///
675 - /// Race-condition note: this reads, the caller adds. Concurrent uploads can
676 - /// produce small overshoots. Acceptable for v1.
677 - /// Row for the storage-cap lookup:
678 - /// (is_internal, enforcement_mode, storage_gb_cap, key_cap, gb_per_key, used_bytes).
679 - type AppStorageCapRow = (bool, String, Option<i32>, Option<i32>, Option<i32>, Option<i64>);
680 -
681 - #[tracing::instrument(skip_all)]
682 - pub async fn would_exceed_storage(
683 - pool: &PgPool,
684 - app_id: SyncAppId,
685 - key: &str,
686 - additional_bytes: i64,
687 - ) -> Result<Option<ExceededLimit>> {
688 - let row: Option<AppStorageCapRow> = sqlx::query_as(
689 - r#"
690 - SELECT sa.is_internal, sa.enforcement_mode,
691 - sa.storage_gb_cap, sa.key_cap, sa.gb_per_key,
692 - u.bytes_stored
693 - FROM sync_apps sa
694 - LEFT JOIN sync_app_usage_current u ON u.app_id = sa.id
695 - WHERE sa.id = $1
696 - "#,
697 - )
698 - .bind(app_id)
699 - .fetch_optional(pool)
700 - .await?;
701 -
702 - let Some((is_internal, mode, storage_gb, key_cap, gb_per_key, bytes_stored)) = row else { return Ok(None); };
703 - if is_internal { return Ok(None); }
704 -
705 - let app_used = bytes_stored.unwrap_or(0);
706 -
707 - match mode.as_str() {
708 - "bulk" => {
709 - let Some(gb) = storage_gb else { return Ok(None) };
710 - let limit = crate::synckit_billing::storage_cap_bytes(gb as u32);
711 - if app_used.saturating_add(additional_bytes) > limit {
712 - return Ok(Some(ExceededLimit {
713 - dimension: "storage",
714 - used: app_used,
715 - limit,
716 - key: None,
717 - }));
718 - }
719 - }
720 - "per_key" => {
721 - let (Some(k), Some(g)) = (key_cap, gb_per_key) else { return Ok(None) };
722 - let per_key_limit = crate::synckit_billing::storage_cap_bytes(g as u32);
723 - let app_limit = crate::synckit_billing::storage_cap_bytes(k.saturating_mul(g) as u32);
724 -
725 - let key_used: Option<i64> = sqlx::query_scalar(
726 - "SELECT bytes_stored FROM sync_key_usage_current
727 - WHERE app_id = $1 AND key = $2",
728 - )
729 - .bind(app_id)
730 - .bind(key)
731 - .fetch_optional(pool)
732 - .await?;
733 - let key_used = key_used.unwrap_or(0);
734 -
735 - if key_used.saturating_add(additional_bytes) > per_key_limit {
736 - return Ok(Some(ExceededLimit {
737 - dimension: "storage_per_key",
738 - used: key_used,
739 - limit: per_key_limit,
740 - key: Some(key.to_string()),
741 - }));
742 - }
743 - // Defensive app-aggregate ceiling. Tripping this before the per-key
744 - // check means the per-key counters have drifted under the app
745 - // counter — refuse rather than admit silent over-allocation.
746 - if app_used.saturating_add(additional_bytes) > app_limit {
747 - return Ok(Some(ExceededLimit {
748 - dimension: "storage",
749 - used: app_used,
750 - limit: app_limit,
751 - key: None,
752 - }));
753 - }
754 - }
755 - _ => return Ok(None),
756 - }
757 -
758 - Ok(None)
759 - }
760 -
761 617 /// Fetch the list of active, non-internal apps that may need a warning email,
762 618 /// along with the data needed to compute which threshold (if any) has been
763 619 /// breached since the last notice. The per-app breach computation is done in
@@ -1007,21 +863,3 @@ pub async fn recalculate_synckit_app_storage(pool: &PgPool) -> Result<u64> {
1007 863 Ok(app_res.rows_affected() + key_res.rows_affected())
1008 864 }
1009 865
1010 - /// Check whether a key is currently actively claimed. Used by the claim
1011 - /// handler to short-circuit the cap check for idempotent re-claims.
1012 - #[tracing::instrument(skip_all)]
1013 - pub async fn is_key_actively_claimed(
1014 - pool: &sqlx::PgPool,
1015 - app_id: SyncAppId,
1016 - key: &str,
1017 - ) -> Result<bool> {
1018 - let row: Option<(uuid::Uuid,)> = sqlx::query_as(
1019 - "SELECT id FROM sync_app_keys
1020 - WHERE app_id = $1 AND key = $2 AND released_at IS NULL",
1021 - )
1022 - .bind(app_id)
1023 - .bind(key)
1024 - .fetch_optional(pool)
1025 - .await?;
1026 - Ok(row.is_some())
1027 - }
M server/src/routes/synckit/blobs.rs +74 -57
@@ -37,7 +37,7 @@ pub(super) async fn blob_upload_url(
37 37 State(state): State<AppState>,
38 38 sync_user: SyncUser,
39 39 Json(req): Json<BlobUploadUrlRequest>,
40 - ) -> Result<impl IntoResponse> {
40 + ) -> Result<Response> {
41 41 let synckit_s3 = state
42 42 .synckit_s3
43 43 .as_ref()
@@ -51,10 +51,17 @@ pub(super) async fn blob_upload_url(
51 51 )));
52 52 }
53 53
54 - // Cap enforcement happens at confirm time (when we know the upload
55 - // succeeded). Presign returns a URL even when the upload would later be
56 - // rejected — cheap, and avoids leaking cap state to unauthenticated
57 - // S3 calls. See blob_confirm_upload for the gate.
54 + // Paid-only gate for first-party apps: don't even hand out an upload URL to
55 + // an unsubscribed user, so they can't stage orphan S3 objects. Non-internal
56 + // (developer-billed) apps are always allowed here. Storage-cap enforcement
57 + // still happens atomically at confirm time.
58 + if !db::synckit::internal_write_allowed(&state.db, sync_user.app_id, sync_user.user_id).await? {
59 + return Ok((
60 + StatusCode::PAYMENT_REQUIRED,
61 + Json(json!({ "reason": "no_subscription" })),
62 + )
63 + .into_response());
64 + }
58 65
59 66 // Check dedup — if this hash already exists, skip upload
60 67 if let Some(_existing) = db::synckit::get_sync_blob_by_hash(
@@ -68,7 +75,8 @@ pub(super) async fn blob_upload_url(
68 75 return Ok(Json(BlobUploadUrlResponse {
69 76 upload_url: String::new(),
70 77 already_exists: true,
71 - }));
78 + })
79 + .into_response());
72 80 }
73 81
74 82 let s3_key = crate::storage::S3Client::generate_synckit_blob_key(
@@ -84,7 +92,10 @@ pub(super) async fn blob_upload_url(
84 92 "application/octet-stream",
85 93 Some(constants::SYNCKIT_BLOB_PRESIGN_EXPIRY_SECS),
86 94 None,
87 - None,
95 + // Bind Content-Length at the S3 layer so the client can't upload
96 + // more than it declared. Confirm reads the actual object size as
97 + // the authoritative figure regardless.
98 + Some(req.size_bytes),
88 99 )
89 100 .await
90 101 .context("presign upload for sync blob")?;
@@ -92,7 +103,8 @@ pub(super) async fn blob_upload_url(
92 103 Ok(Json(BlobUploadUrlResponse {
93 104 upload_url,
94 105 already_exists: false,
95 - }))
106 + })
107 + .into_response())
96 108 }
97 109
98 110 /// Confirm that a blob upload to S3 completed successfully.
@@ -117,12 +129,41 @@ pub(super) async fn blob_confirm_upload(
117 129
118 130 validation::validate_sync_blob_hash(&req.hash)?;
119 131
120 - // Billing + cap enforcement (skipped entirely for internal apps).
121 132 let billing = synckit_billing::get_app_with_billing(&state.db, sync_user.app_id)
122 133 .await?
123 134 .ok_or(AppError::NotFound)?;
124 135
125 - if !billing.is_internal {
136 + let s3_key = crate::storage::S3Client::generate_synckit_blob_key(
137 + sync_user.app_id, sync_user.user_id, &req.hash,
138 + );
139 +
140 + // The authoritative size is the actual S3 object, never the client's
141 + // claim. `object_size` doubles as the existence check (None = not there).
142 + let actual_size = synckit_s3
143 + .object_size(&s3_key)
144 + .await?
145 + .ok_or_else(|| {
146 + AppError::BadRequest("Blob not found in storage — upload before confirming".to_string())
147 + })?;
148 + if actual_size <= 0 || actual_size > constants::SYNCKIT_MAX_BLOB_SIZE_BYTES {
149 + return Err(AppError::BadRequest(format!(
150 + "Stored blob size {actual_size} is out of range"
151 + )));
152 + }
153 +
154 + // Clear the pending upload record now that the upload is confirmed.
155 + db::pending_uploads::remove_pending_upload(&state.db, sync_user.user_id, &s3_key, "synckit").await?;
156 +
157 + // Record the blob and enforce the cap atomically. Internal apps gate on the
158 + // user's paid subscription + per-user cap; developer apps on the app/per-key
159 + // counters. Both are single-transaction (lock, check, insert, count).
160 + let outcome = if billing.is_internal {
161 + db::synckit::confirm_internal_blob(
162 + &state.db, sync_user.app_id, sync_user.user_id,
163 + &req.hash, actual_size, &s3_key, &sync_user.key,
164 + )
165 + .await?
166 + } else {
126 167 if billing.billing_status != "active" {
127 168 return Ok((
128 169 StatusCode::PAYMENT_REQUIRED,
@@ -130,60 +171,36 @@ pub(super) async fn blob_confirm_upload(
130 171 )
131 172 .into_response());
132 173 }
133 - if let Some(exceeded) = synckit_billing::would_exceed_storage(
134 - &state.db, sync_user.app_id, &sync_user.key, req.size_bytes,
135 - ).await? {
174 + db::synckit::confirm_developer_blob(
175 + &state.db, sync_user.app_id, sync_user.user_id,
176 + &req.hash, actual_size, &s3_key, &sync_user.key,
177 + &billing.enforcement_mode, billing.storage_gb_cap, billing.key_cap, billing.gb_per_key,
178 + )
179 + .await?
180 + };
181 +
182 + match outcome {
183 + db::synckit::BlobConfirm::Stored | db::synckit::BlobConfirm::AlreadyStored => {
184 + Ok(StatusCode::NO_CONTENT.into_response())
185 + }
186 + db::synckit::BlobConfirm::NoSubscription => Ok((
187 + StatusCode::PAYMENT_REQUIRED,
188 + Json(json!({ "reason": "no_subscription" })),
189 + )
190 + .into_response()),
191 + db::synckit::BlobConfirm::QuotaExceeded { dimension, used, limit, key } => {
136 192 let mut body = json!({
137 193 "reason": "storage_limit_reached",
138 - "dimension": exceeded.dimension,
139 - "used": exceeded.used,
140 - "limit": exceeded.limit,
194 + "dimension": dimension,
195 + "used": used,
196 + "limit": limit,
141 197 });
142 - if let Some(k) = &exceeded.key {
198 + if let Some(k) = key {
143 199 body["key"] = json!(k);
144 200 }
145 - return Ok((StatusCode::PAYMENT_REQUIRED, Json(body)).into_response());
201 + Ok((StatusCode::PAYMENT_REQUIRED, Json(body)).into_response())
146 202 }
147 203 }
148 -
149 - let s3_key = crate::storage::S3Client::generate_synckit_blob_key(
150 - sync_user.app_id, sync_user.user_id, &req.hash,
151 - );
152 -
153 - // Verify the object actually exists in S3
154 - if !synckit_s3.object_exists(&s3_key).await? {
155 - return Err(AppError::BadRequest(
156 - "Blob not found in storage — upload before confirming".to_string(),
157 - ));
158 - }
159 -
160 - // Clear the pending upload record now that the upload is confirmed
161 - db::pending_uploads::remove_pending_upload(&state.db, sync_user.user_id, &s3_key, "synckit").await?;
162 -
163 - // Record in DB (idempotent — ON CONFLICT DO NOTHING if already exists)
164 - db::synckit::create_sync_blob_idempotent(
165 - &state.db,
166 - sync_user.app_id,
167 - sync_user.user_id,
168 - &req.hash,
169 - req.size_bytes,
170 - &s3_key,
171 - &sync_user.key,
172 - )
173 - .await?;
174 -
175 - // Update the rolling storage counter (app-level + per-key). We don't fail
176 - // the request if this breaks — the weekly drift correction job will
177 - // reconcile from sync_blobs. Skip for internal apps to keep the counter
178 - // at 0 there.
179 - if !billing.is_internal
180 - && let Err(e) = synckit_billing::add_bytes_stored(
181 - &state.db, sync_user.app_id, &sync_user.key, req.size_bytes,
182 - ).await {
183 - tracing::error!(error = ?e, app_id = %sync_user.app_id, "failed to bump bytes_stored");
184 - }
185 -
186 - Ok(StatusCode::NO_CONTENT.into_response())
187 204 }
188 205
189 206 /// Request a pre-signed S3 download URL for a blob by hash.
M server/src/routes/synckit/keys.rs +22 -25
@@ -51,7 +51,10 @@ pub(super) async fn claim(
51 51 .await?
52 52 .ok_or(AppError::NotFound)?;
53 53
54 - if !billing.is_internal {
54 + // `key_cap` is enforced inside `claim_key`, under the usage-row lock, so
55 + // concurrent claims of distinct keys can't over-allocate. `None` means
56 + // uncapped (internal apps, or `bulk` developer apps).
57 + let key_cap = if !billing.is_internal {
55 58 if billing.billing_status != "active" {
56 59 return Ok((
57 60 StatusCode::PAYMENT_REQUIRED,
@@ -59,33 +62,27 @@ pub(super) async fn claim(
59 62 )
60 63 .into_response());
61 64 }
62 -
63 65 if billing.enforcement_mode == "per_key" {
64 - let key_cap = billing.key_cap.unwrap_or(0);
65 - let keys_claimed = billing.keys_claimed.unwrap_or(0);
66 - if keys_claimed >= key_cap {
67 - // Re-claim of an already-active key is always OK — it doesn't
68 - // consume a new slot.
69 - let already_active = synckit_billing::is_key_actively_claimed(
70 - &state.db, app.id, &req.key,
71 - )
72 - .await?;
73 - if !already_active {
74 - return Ok((
75 - StatusCode::PAYMENT_REQUIRED,
76 - Json(json!({
77 - "reason": "key_limit_reached",
78 - "key_cap": key_cap,
79 - "keys_claimed": keys_claimed,
80 - })),
81 - )
82 - .into_response());
83 - }
84 - }
66 + Some(billing.key_cap.unwrap_or(0))
67 + } else {
68 + None
85 69 }
86 - }
70 + } else {
71 + None
72 + };
87 73
88 - let result = synckit_billing::claim_key(&state.db, app.id, &req.key).await?;
74 + let result = synckit_billing::claim_key(&state.db, app.id, &req.key, key_cap).await?;
75 + if result.cap_reached {
76 + return Ok((
77 + StatusCode::PAYMENT_REQUIRED,
78 + Json(json!({
79 + "reason": "key_limit_reached",
80 + "key_cap": key_cap.unwrap_or(0),
81 + "keys_claimed": result.total_claimed,
82 + })),
83 + )
84 + .into_response());
85 + }
89 86 Ok(Json(ClaimKeyResponse {
90 87 newly_claimed: result.newly_claimed,
91 88 total_claimed: result.total_claimed,
M server/src/routes/synckit/mod.rs +3 -1
@@ -369,7 +369,9 @@ pub(crate) struct BlobUploadUrlResponse {
369 369 #[derive(Deserialize, utoipa::ToSchema)]
370 370 pub(crate) struct BlobConfirmRequest {
371 371 pub hash: String,
372 - pub size_bytes: i64,
372 + // The confirm handler reads the authoritative object size from S3 and does
373 + // not trust a client-declared size. Clients may still send `size_bytes`; it
374 + // is ignored by deserialization (no `deny_unknown_fields`).
373 375 }
374 376
375 377 #[derive(Deserialize, utoipa::ToSchema)]
M server/src/routes/synckit/sync.rs +16 -3
@@ -2,9 +2,11 @@
2 2
3 3 use axum::{
4 4 extract::{Path, State},
5 - response::IntoResponse,
5 + http::StatusCode,
6 + response::{IntoResponse, Response},
6 7 Json,
7 8 };
9 + use serde_json::json;
8 10
9 11 use crate::{
10 12 constants,
@@ -39,12 +41,23 @@ pub(super) async fn sync_push(
39 41 State(state): State<AppState>,
40 42 sync_user: SyncUser,
41 43 Json(req): Json<PushRequest>,
42 - ) -> Result<impl IntoResponse> {
44 + ) -> Result<Response> {
43 45 let app_id = sync_user.app_id;
44 46 let user_id = sync_user.user_id;
45 47 tracing::Span::current().record("app_id", tracing::field::display(&app_id));
46 48 tracing::Span::current().record("user_id", tracing::field::display(&user_id));
47 49
50 + // Paid-only sync: first-party apps require an active end-user subscription
51 + // to write. Reads (pull) stay open so a lapsed user can still export their
52 + // data. Non-internal (developer-billed) apps are always allowed here.
53 + if !db::synckit::internal_write_allowed(&state.db, app_id, user_id).await? {
54 + return Ok((
55 + StatusCode::PAYMENT_REQUIRED,
56 + Json(json!({ "reason": "no_subscription" })),
57 + )
58 + .into_response());
59 + }
60 +
48 61 if req.changes.is_empty() {
49 62 return Err(AppError::BadRequest("No changes provided".to_string()));
50 63 }
@@ -105,7 +118,7 @@ pub(super) async fn sync_push(
105 118 let _ = sender.send(()); // Ignore errors (no subscribers = ok)
106 119 }
107 120
108 - Ok(Json(PushResponse { cursor }))
121 + Ok(Json(PushResponse { cursor }).into_response())
109 122 }
110 123
111 124 /// Pull changelog entries after a given cursor.
M server/tests/harness/storage.rs +4
@@ -64,6 +64,10 @@ impl StorageBackend for InMemoryStorage {
64 64 .ok_or_else(|| AppError::Storage(format!("Object not found: {}", s3_key)))
65 65 }
66 66
67 + async fn download_object_buf(&self, s3_key: &str) -> Result<bytes::Bytes> {
68 + self.download_object(s3_key).await.map(bytes::Bytes::from)
69 + }
70 +
67 71 async fn download_stream(&self, s3_key: &str) -> Result<s3_storage::ByteStream> {
68 72 let bytes = self
69 73 .objects
M server/tests/workflows/mod.rs +1
@@ -17,6 +17,7 @@ mod subscriptions;
17 17 mod synckit;
18 18 mod synckit_billing;
19 19 mod synckit_per_key_storage;
20 + mod synckit_paid_sync;
20 21 mod synckit_adversarial;
21 22 mod pages;
22 23 mod analytics;
M server/tests/workflows/synckit_adversarial.rs +3 -1
@@ -124,7 +124,9 @@ async fn adversarial_jwt_key_with_sql_injection_literal() {
124 124 auth_as(&mut h, user_id, app_id, evil);
125 125 let hash = fake_hash(0x01);
126 126 let s3_key = format!("{}/{}/{}", app_id, user_id, &hash);
127 - blobs.put(&s3_key, vec![0u8; 8]);
127 + // Confirm records the authoritative S3 object size, so store the full
128 + // declared count to keep the per-key counter assertion below meaningful.
129 + blobs.put(&s3_key, vec![0u8; 1024]);
128 130
129 131 let r = h
130 132 .client
A server/tests/workflows/synckit_paid_sync.rs +227
@@ -0,0 +1,227 @@
1 + //! Paid-only first-party (internal) SyncKit: a logged-in user of an internal
2 + //! app (GO/BB/AF) may sync only with an `active` end-user subscription, and the
3 + //! per-user `storage_limit_bytes` is enforced atomically from the authoritative
4 + //! S3 object size (not the client's declared size). Closes ultra-fuzz Run 3
5 + //! CRITICAL #1 (caps never enforced for internal apps; no sub required) and
6 + //! SERIOUS #2 (confirm trusted the client byte count).
7 +
8 + use std::sync::Arc;
9 +
10 + use makenotwork::db::{SyncAppId, UserId};
11 + use serde_json::json;
12 + use sqlx::PgPool;
13 +
14 + use crate::harness::{client::TestResponse, storage::InMemoryStorage, stripe, BuildOptions, TestHarness};
15 +
16 + const GIB: i64 = 1024 * 1024 * 1024;
17 +
18 + async fn harness_with_blobs() -> (TestHarness, Arc<InMemoryStorage>) {
19 + let synckit_mem = Arc::new(InMemoryStorage::new());
20 + let mock_stripe = Arc::new(stripe::MockPaymentProvider::new());
21 + let mut h = TestHarness::build(BuildOptions {
22 + synckit_storage: Some(synckit_mem.clone()),
23 + stripe_client: Some(mock_stripe.clone()),
24 + ..Default::default()
25 + })
26 + .await;
27 + h.mock_stripe = Some(mock_stripe);
28 + (h, synckit_mem)
29 + }
30 +
31 + /// Create an active first-party (internal) sync app.
32 + async fn create_internal_app(pool: &PgPool, user_id: UserId) -> (SyncAppId, String) {
33 + let api_key = "test-api-key-paid-sync";
34 + let key_hash = crate::harness::hash_api_key(api_key);
35 + let key_prefix = &api_key[..8];
36 + let app_id: SyncAppId = sqlx::query_scalar(
37 + "INSERT INTO sync_apps (creator_id, name, api_key_hash, api_key_prefix, is_internal, billing_status)
38 + VALUES ($1, 'GoingsOn', $2, $3, TRUE, 'active')
39 + RETURNING id",
40 + )
41 + .bind(user_id)
42 + .bind(&key_hash)
43 + .bind(key_prefix)
44 + .fetch_one(pool)
45 + .await
46 + .expect("insert internal sync_app");
47 + (app_id, api_key.to_string())
48 + }
49 +
50 + /// Seed an end-user subscription row with the given status and cap.
51 + async fn seed_subscription(pool: &PgPool, user_id: UserId, app_id: SyncAppId, status: &str, limit_bytes: i64) {
52 + sqlx::query(
53 + "INSERT INTO app_sync_subscriptions
54 + (user_id, app_id, stripe_subscription_id, stripe_customer_id, tier, status, storage_limit_bytes)
55 + VALUES ($1, $2, $3, 'cus_test', 'monthly', $4, $5)",
56 + )
57 + .bind(user_id)
58 + .bind(app_id)
59 + .bind(format!("sub_{user_id}_{app_id}"))
60 + .bind(status)
61 + .bind(limit_bytes)
62 + .execute(pool)
63 + .await
64 + .expect("seed subscription");
65 + }
66 +
67 + fn auth_as(h: &mut TestHarness, user_id: UserId, app_id: SyncAppId, key: &str) {
68 + let token = makenotwork::synckit_auth::create_sync_token(
69 + "test-synckit-jwt-secret",
70 + user_id,
71 + app_id,
72 + key,
73 + )
74 + .expect("mint test JWT");
75 + h.client.set_bearer_token(&token);
76 + }
77 +
78 + fn fake_hash(seed: u8) -> String {
79 + let mut s = String::with_capacity(64);
80 + for _ in 0..32 {
81 + s.push_str(&format!("{seed:02x}"));
82 + }
83 + s
84 + }
85 +
86 + /// Put a blob of `actual_bytes` directly into storage, then call confirm with
87 + /// the (possibly different) `declared_bytes`. Returns the confirm response.
88 + async fn put_and_confirm(
89 + h: &mut TestHarness,
90 + blobs: &Arc<InMemoryStorage>,
91 + app_id: SyncAppId,
92 + user_id: UserId,
93 + hash: &str,
94 + actual_bytes: usize,
95 + declared_bytes: i64,
96 + ) -> TestResponse {
97 + let s3_key = format!("{app_id}/{user_id}/{hash}");
98 + blobs.put(&s3_key, vec![0u8; actual_bytes]);
99 + h.client
100 + .post_json(
101 + "/api/sync/blobs/confirm",
102 + &json!({ "hash": hash, "size_bytes": declared_bytes }).to_string(),
103 + )
104 + .await
105 + }
106 +
107 + // ── Tests ──
108 +
109 + #[tokio::test]
110 + async fn upload_url_refused_without_subscription() {
111 + let (mut h, _blobs) = harness_with_blobs().await;
112 + let user_id = h.signup("paid_nosub", "paid_nosub@example.com", "Password1!").await;
113 + let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
114 + auth_as(&mut h, user_id, app_id, "user-key");
115 +
116 + let resp = h
117 + .client
118 + .post_json(
119 + "/api/sync/blobs/upload",
120 + &json!({ "hash": fake_hash(0x01), "size_bytes": 100 }).to_string(),
121 + )
122 + .await;
123 + assert_eq!(resp.status, 402, "upload-url must be refused without a sub: {}", resp.text);
124 + let body: serde_json::Value = resp.json();
125 + assert_eq!(body["reason"], "no_subscription");
126 + }
127 +
128 + #[tokio::test]
129 + async fn confirm_refused_without_subscription() {
130 + let (mut h, blobs) = harness_with_blobs().await;
131 + let user_id = h.signup("paid_nosub2", "paid_nosub2@example.com", "Password1!").await;
132 + let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
133 + auth_as(&mut h, user_id, app_id, "user-key");
134 +
135 + // Bypass the upload-url gate and put the object directly to exercise the
136 + // confirm-side subscription check.
137 + let resp = put_and_confirm(&mut h, &blobs, app_id, user_id, &fake_hash(0x02), 100, 100).await;
138 + assert_eq!(resp.status, 402, "confirm must be refused without a sub: {}", resp.text);
139 + let body: serde_json::Value = resp.json();
140 + assert_eq!(body["reason"], "no_subscription");
141 + }
142 +
143 + #[tokio::test]
144 + async fn push_refused_without_subscription() {
145 + let (mut h, _blobs) = harness_with_blobs().await;
146 + let user_id = h.signup("paid_push", "paid_push@example.com", "Password1!").await;
147 + let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
148 + auth_as(&mut h, user_id, app_id, "user-key");
149 +
150 + let resp = h
151 + .client
152 + .post_json(
153 + "/api/sync/push",
154 + &json!({
155 + "device_id": uuid::Uuid::new_v4().to_string(),
156 + "batch_id": uuid::Uuid::new_v4().to_string(),
157 + "changes": [
158 + { "table": "tasks", "op": "INSERT", "row_id": "a", "timestamp": "2025-01-01T00:00:00Z", "data": {"t": 1} }
159 + ]
160 + })
161 + .to_string(),
162 + )
163 + .await;
164 + assert_eq!(resp.status, 402, "push must be refused without a sub: {}", resp.text);
165 + let body: serde_json::Value = resp.json();
166 + assert_eq!(body["reason"], "no_subscription");
167 + }
168 +
169 + #[tokio::test]
170 + async fn confirm_succeeds_within_cap_with_active_subscription() {
171 + let (mut h, blobs) = harness_with_blobs().await;
172 + let user_id = h.signup("paid_ok", "paid_ok@example.com", "Password1!").await;
173 + let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
174 + seed_subscription(&h.db, user_id, app_id, "active", GIB).await;
175 + auth_as(&mut h, user_id, app_id, "user-key");
176 +
177 + let resp = put_and_confirm(&mut h, &blobs, app_id, user_id, &fake_hash(0x03), 1000, 1000).await;
178 + assert_eq!(resp.status, 204, "confirm within cap should succeed: {}", resp.text);
179 +
180 + // The blob is recorded with the authoritative size.
181 + let stored: i64 = sqlx::query_scalar(
182 + "SELECT size_bytes FROM sync_blobs WHERE app_id = $1 AND user_id = $2 AND hash = $3",
183 + )
184 + .bind(app_id)
185 + .bind(user_id)
186 + .bind(fake_hash(0x03))
187 + .fetch_one(&h.db)
188 + .await
189 + .unwrap();
190 + assert_eq!(stored, 1000);
191 + }
192 +
193 + #[tokio::test]
194 + async fn confirm_rejected_over_cap() {
195 + let (mut h, blobs) = harness_with_blobs().await;
196 + let user_id = h.signup("paid_over", "paid_over@example.com", "Password1!").await;
197 + let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
198 + seed_subscription(&h.db, user_id, app_id, "active", 50).await; // 50-byte cap
199 + auth_as(&mut h, user_id, app_id, "user-key");
200 +
201 + let resp = put_and_confirm(&mut h, &blobs, app_id, user_id, &fake_hash(0x04), 100, 100).await;
202 + assert_eq!(resp.status, 402, "confirm over cap should be refused: {}", resp.text);
203 + let body: serde_json::Value = resp.json();
204 + assert_eq!(body["reason"], "storage_limit_reached");
205 + assert_eq!(body["dimension"], "storage");
206 + assert_eq!(body["limit"], 50);
207 + }
208 +
209 + #[tokio::test]
210 + async fn confirm_uses_authoritative_object_size_not_client_claim() {
211 + let (mut h, blobs) = harness_with_blobs().await;
212 + let user_id = h.signup("paid_spoof", "paid_spoof@example.com", "Password1!").await;
213 + let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
214 + seed_subscription(&h.db, user_id, app_id, "active", 50).await; // 50-byte cap
215 + auth_as(&mut h, user_id, app_id, "user-key");
216 +
217 + // Client lies: declares 1 byte but the actual S3 object is 100 bytes. The
218 + // cap (50) must be enforced against the real 100, not the claimed 1.
219 + let resp = put_and_confirm(&mut h, &blobs, app_id, user_id, &fake_hash(0x05), 100, 1).await;
220 + assert_eq!(
221 + resp.status, 402,
222 + "size spoof (declare 1, upload 100) must be caught by object_size: {}",
223 + resp.text
224 + );
225 + let body: serde_json::Value = resp.json();
226 + assert_eq!(body["reason"], "storage_limit_reached");
227 + }
M server/tests/workflows/synckit_per_key_storage.rs +3 -1
@@ -152,8 +152,10 @@ async fn upload_and_confirm(
152 152 assert_eq!(resp.status, 200, "blob upload-url failed: {}", resp.text);
153 153
154 154 // Mimic the client PUTting to the presigned URL by populating storage.
155 + // Store the full byte count: confirm now reads the authoritative S3 object
156 + // size (not the client-declared size), so the stored object must match.
155 157 let s3_key = format!("{}/{}/{}", app_id, user_id, hash);
156 - blobs.put(&s3_key, vec![0u8; size_bytes.min(8) as usize]);
158 + blobs.put(&s3_key, vec![0u8; size_bytes as usize]);
157 159
158 160 h.client
159 161 .post_json(