Skip to main content

max / makenotwork

15.1 KB · 459 lines History Blame Raw
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::fmt::Write as _;
9 use std::sync::Arc;
10
11 use makenotwork::db::{SyncAppId, UserId};
12 use serde_json::json;
13 use sqlx::PgPool;
14
15 use crate::harness::{
16 BuildOptions, TestHarness, client::TestResponse, storage::InMemoryStorage, stripe,
17 };
18
19 const GIB: i64 = 1024 * 1024 * 1024;
20
21 pub(crate) async fn harness_with_blobs() -> (TestHarness, Arc<InMemoryStorage>) {
22 let synckit_mem = Arc::new(InMemoryStorage::new());
23 let mock_stripe = Arc::new(stripe::MockPaymentProvider::new());
24 let mut h = TestHarness::build(BuildOptions {
25 synckit_storage: Some(synckit_mem.clone()),
26 stripe_client: Some(mock_stripe.clone()),
27 ..Default::default()
28 })
29 .await;
30 h.mock_stripe = Some(mock_stripe);
31 (h, synckit_mem)
32 }
33
34 /// Create an active first-party (internal) sync app.
35 pub(crate) async fn create_internal_app(pool: &PgPool, user_id: UserId) -> (SyncAppId, String) {
36 let api_key = "test-api-key-paid-sync";
37 let key_hash = crate::harness::hash_api_key(api_key);
38 let key_prefix = &api_key[..8];
39 let app_id: SyncAppId = sqlx::query_scalar(
40 "INSERT INTO sync_apps (creator_id, name, api_key_hash, api_key_prefix, is_internal, billing_status)
41 VALUES ($1, 'GoingsOn', $2, $3, TRUE, 'active')
42 RETURNING id",
43 )
44 .bind(user_id)
45 .bind(&key_hash)
46 .bind(key_prefix)
47 .fetch_one(pool)
48 .await
49 .expect("insert internal sync_app");
50 (app_id, api_key.to_string())
51 }
52
53 /// Seed an end-user subscription row with the given status and cap.
54 pub(crate) async fn seed_subscription(
55 pool: &PgPool,
56 user_id: UserId,
57 app_id: SyncAppId,
58 status: &str,
59 limit_bytes: i64,
60 ) {
61 sqlx::query(
62 "INSERT INTO app_sync_subscriptions
63 (user_id, app_id, stripe_subscription_id, stripe_customer_id, tier, status, storage_limit_bytes)
64 VALUES ($1, $2, $3, 'cus_test', 'monthly', $4, $5)",
65 )
66 .bind(user_id)
67 .bind(app_id)
68 .bind(format!("sub_{user_id}_{app_id}"))
69 .bind(status)
70 .bind(limit_bytes)
71 .execute(pool)
72 .await
73 .expect("seed subscription");
74 }
75
76 pub(crate) fn auth_as(h: &mut TestHarness, user_id: UserId, app_id: SyncAppId, key: &str) {
77 let token = makenotwork::synckit_auth::create_sync_token(
78 "test-synckit-jwt-secret",
79 user_id,
80 app_id,
81 key,
82 )
83 .expect("mint test JWT");
84 h.client.set_bearer_token(&token);
85 }
86
87 pub(crate) fn fake_hash(seed: u8) -> String {
88 let mut s = String::with_capacity(64);
89 for _ in 0..32 {
90 write!(s, "{seed:02x}").unwrap();
91 }
92 s
93 }
94
95 /// Put a blob of `actual_bytes` directly into storage, then call confirm with
96 /// the (possibly different) `declared_bytes`. Returns the confirm response.
97 async fn put_and_confirm(
98 h: &mut TestHarness,
99 blobs: &Arc<InMemoryStorage>,
100 app_id: SyncAppId,
101 user_id: UserId,
102 hash: &str,
103 actual_bytes: usize,
104 declared_bytes: i64,
105 ) -> TestResponse {
106 let s3_key = format!("{app_id}/{user_id}/{hash}");
107 blobs.put(&s3_key, vec![0u8; actual_bytes]);
108 h.client
109 .post_json(
110 "/api/sync/blobs/confirm",
111 &json!({ "hash": hash, "size_bytes": declared_bytes }).to_string(),
112 )
113 .await
114 }
115
116 // ── Tests ──
117
118 #[tokio::test]
119 async fn upload_url_refused_without_subscription() {
120 let (mut h, _blobs) = harness_with_blobs().await;
121 let user_id = h
122 .signup("paid_nosub", "paid_nosub@example.com", "Password1!")
123 .await;
124 let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
125 auth_as(&mut h, user_id, app_id, "user-key");
126
127 let resp = h
128 .client
129 .post_json(
130 "/api/sync/blobs/upload",
131 &json!({ "hash": fake_hash(0x01), "size_bytes": 100 }).to_string(),
132 )
133 .await;
134 assert_eq!(
135 resp.status, 402,
136 "upload-url must be refused without a sub: {}",
137 resp.text
138 );
139 let body: serde_json::Value = resp.json();
140 assert_eq!(body["reason"], "no_subscription");
141 }
142
143 #[tokio::test]
144 async fn confirm_refused_without_subscription() {
145 let (mut h, blobs) = harness_with_blobs().await;
146 let user_id = h
147 .signup("paid_nosub2", "paid_nosub2@example.com", "Password1!")
148 .await;
149 let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
150 auth_as(&mut h, user_id, app_id, "user-key");
151
152 // Bypass the upload-url gate and put the object directly to exercise the
153 // confirm-side subscription check.
154 let resp = put_and_confirm(&mut h, &blobs, app_id, user_id, &fake_hash(0x02), 100, 100).await;
155 assert_eq!(
156 resp.status, 402,
157 "confirm must be refused without a sub: {}",
158 resp.text
159 );
160 let body: serde_json::Value = resp.json();
161 assert_eq!(body["reason"], "no_subscription");
162 }
163
164 #[tokio::test]
165 async fn push_refused_without_subscription() {
166 let (mut h, _blobs) = harness_with_blobs().await;
167 let user_id = h
168 .signup("paid_push", "paid_push@example.com", "Password1!")
169 .await;
170 let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
171 auth_as(&mut h, user_id, app_id, "user-key");
172
173 let resp = h
174 .client
175 .post_json(
176 "/api/sync/push",
177 &json!({
178 "device_id": uuid::Uuid::new_v4().to_string(),
179 "batch_id": uuid::Uuid::new_v4().to_string(),
180 "changes": [
181 { "table": "tasks", "op": "INSERT", "row_id": "a", "timestamp": "2025-01-01T00:00:00Z", "data": {"t": 1} }
182 ]
183 })
184 .to_string(),
185 )
186 .await;
187 assert_eq!(
188 resp.status, 402,
189 "push must be refused without a sub: {}",
190 resp.text
191 );
192 let body: serde_json::Value = resp.json();
193 assert_eq!(body["reason"], "no_subscription");
194 }
195
196 #[tokio::test]
197 async fn confirm_succeeds_within_cap_with_active_subscription() {
198 let (mut h, blobs) = harness_with_blobs().await;
199 let user_id = h
200 .signup("paid_ok", "paid_ok@example.com", "Password1!")
201 .await;
202 let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
203 seed_subscription(&h.db, user_id, app_id, "active", GIB).await;
204 auth_as(&mut h, user_id, app_id, "user-key");
205
206 let resp = put_and_confirm(
207 &mut h,
208 &blobs,
209 app_id,
210 user_id,
211 &fake_hash(0x03),
212 1000,
213 1000,
214 )
215 .await;
216 assert_eq!(
217 resp.status, 204,
218 "confirm within cap should succeed: {}",
219 resp.text
220 );
221
222 // The blob is recorded with the authoritative size.
223 let stored: i64 = sqlx::query_scalar(
224 "SELECT size_bytes FROM sync_blobs WHERE app_id = $1 AND user_id = $2 AND hash = $3",
225 )
226 .bind(app_id)
227 .bind(user_id)
228 .bind(fake_hash(0x03))
229 .fetch_one(&h.db)
230 .await
231 .unwrap();
232 assert_eq!(stored, 1000);
233 }
234
235 #[tokio::test]
236 async fn confirm_rejected_over_cap() {
237 let (mut h, blobs) = harness_with_blobs().await;
238 let user_id = h
239 .signup("paid_over", "paid_over@example.com", "Password1!")
240 .await;
241 let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
242 seed_subscription(&h.db, user_id, app_id, "active", 50).await; // 50-byte cap
243 auth_as(&mut h, user_id, app_id, "user-key");
244
245 let resp = put_and_confirm(&mut h, &blobs, app_id, user_id, &fake_hash(0x04), 100, 100).await;
246 assert_eq!(
247 resp.status, 402,
248 "confirm over cap should be refused: {}",
249 resp.text
250 );
251 let body: serde_json::Value = resp.json();
252 assert_eq!(body["reason"], "storage_limit_reached");
253 assert_eq!(body["dimension"], "storage");
254 assert_eq!(body["limit"], 50);
255 }
256
257 #[tokio::test]
258 async fn confirm_uses_authoritative_object_size_not_client_claim() {
259 let (mut h, blobs) = harness_with_blobs().await;
260 let user_id = h
261 .signup("paid_spoof", "paid_spoof@example.com", "Password1!")
262 .await;
263 let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
264 seed_subscription(&h.db, user_id, app_id, "active", 50).await; // 50-byte cap
265 auth_as(&mut h, user_id, app_id, "user-key");
266
267 // Client lies: declares 1 byte but the actual S3 object is 100 bytes. The
268 // cap (50) must be enforced against the real 100, not the claimed 1.
269 let resp = put_and_confirm(&mut h, &blobs, app_id, user_id, &fake_hash(0x05), 100, 1).await;
270 assert_eq!(
271 resp.status, 402,
272 "size spoof (declare 1, upload 100) must be caught by object_size: {}",
273 resp.text
274 );
275 let body: serde_json::Value = resp.json();
276 assert_eq!(body["reason"], "storage_limit_reached");
277 }
278
279 #[tokio::test]
280 async fn delete_blob_removes_row_dead_letters_object_and_frees_space() {
281 let (mut h, blobs) = harness_with_blobs().await;
282 let user_id = h
283 .signup("paid_del", "paid_del@example.com", "Password1!")
284 .await;
285 let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
286 seed_subscription(&h.db, user_id, app_id, "active", 1500).await; // 1500-byte cap
287 auth_as(&mut h, user_id, app_id, "user-key");
288
289 // Fill most of the cap with one 1000-byte blob.
290 let hash = fake_hash(0x06);
291 let resp = put_and_confirm(&mut h, &blobs, app_id, user_id, &hash, 1000, 1000).await;
292 assert_eq!(
293 resp.status, 204,
294 "first confirm should succeed: {}",
295 resp.text
296 );
297
298 // A second 1000-byte blob would breach the 1500 cap (1000 + 1000 > 1500).
299 let hash2 = fake_hash(0x07);
300 let resp = put_and_confirm(&mut h, &blobs, app_id, user_id, &hash2, 1000, 1000).await;
301 assert_eq!(
302 resp.status, 402,
303 "second confirm should breach the cap: {}",
304 resp.text
305 );
306
307 // Delete the first blob.
308 let s3_key = format!("{app_id}/{user_id}/{hash}");
309 let resp = h.client.delete(&format!("/api/sync/blobs/{hash}")).await;
310 assert_eq!(resp.status, 204, "delete should succeed: {}", resp.text);
311
312 // Row is gone.
313 let remaining: i64 = sqlx::query_scalar(
314 "SELECT COUNT(*) FROM sync_blobs WHERE app_id = $1 AND user_id = $2 AND hash = $3",
315 )
316 .bind(app_id)
317 .bind(user_id)
318 .bind(&hash)
319 .fetch_one(&h.db)
320 .await
321 .unwrap();
322 assert_eq!(remaining, 0, "deleted blob row must be gone");
323
324 // The S3 object was dead-lettered for the synckit bucket, not dropped.
325 let dead_lettered: i64 = sqlx::query_scalar(
326 "SELECT COUNT(*) FROM pending_s3_deletions
327 WHERE s3_key = $1 AND bucket = 'synckit' AND source = 'synckit_blob_delete'",
328 )
329 .bind(&s3_key)
330 .fetch_one(&h.db)
331 .await
332 .unwrap();
333 assert_eq!(
334 dead_lettered, 1,
335 "deleted object must be enqueued to the dead-letter ladder"
336 );
337
338 // Space is freed: the previously-rejected second blob now fits (internal
339 // usage is summed from sync_blobs, so removing the row shrinks usage).
340 let resp = put_and_confirm(&mut h, &blobs, app_id, user_id, &hash2, 1000, 1000).await;
341 assert_eq!(
342 resp.status, 204,
343 "after delete, the second blob should fit: {}",
344 resp.text
345 );
346 }
347
348 #[tokio::test]
349 async fn compaction_waits_for_a_freshly_registered_device() {
350 // Regression: a new device sits at last_pulled_seq = 0 until its first pull.
351 // Compaction must NOT delete log entries below the other devices' cursors
352 // while a never-pulled device exists, or that device's first pull-from-0
353 // would silently miss the compacted-away changes.
354 let (mut h, _blobs) = harness_with_blobs().await;
355 let user_id = h
356 .signup("compact_dev", "compact_dev@example.com", "Password1!")
357 .await;
358 let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
359
360 // Device A has pulled; device B is freshly registered (seq 0).
361 let dev_a: uuid::Uuid = sqlx::query_scalar(
362 "INSERT INTO sync_devices (app_id, user_id, device_name, platform)
363 VALUES ($1, $2, 'A', 'test') RETURNING id",
364 )
365 .bind(app_id)
366 .bind(user_id)
367 .fetch_one(&h.db)
368 .await
369 .unwrap();
370 let _dev_b: uuid::Uuid = sqlx::query_scalar(
371 "INSERT INTO sync_devices (app_id, user_id, device_name, platform)
372 VALUES ($1, $2, 'B', 'test') RETURNING id",
373 )
374 .bind(app_id)
375 .bind(user_id)
376 .fetch_one(&h.db)
377 .await
378 .unwrap();
379
380 // One old log entry (backdated well past the 7-day compaction margin).
381 let seq: i64 = sqlx::query_scalar(
382 "INSERT INTO sync_log
383 (app_id, user_id, device_id, table_name, operation, row_id, client_timestamp, created_at)
384 VALUES ($1, $2, $3, 'tasks', 'INSERT', 'r1', NOW(), NOW() - INTERVAL '30 days')
385 RETURNING seq",
386 )
387 .bind(app_id).bind(user_id).bind(dev_a).fetch_one(&h.db).await.unwrap();
388
389 // Device A has pulled past the entry; device B has not pulled at all.
390 sqlx::query("UPDATE sync_devices SET last_pulled_seq = $1 WHERE id = $2")
391 .bind(seq)
392 .bind(dev_a)
393 .execute(&h.db)
394 .await
395 .unwrap();
396
397 // With device B pinned at 0, compaction must delete nothing.
398 let deleted = makenotwork::db::synckit::compact_all_sync_logs(&h.db, 7)
399 .await
400 .unwrap();
401 assert_eq!(
402 deleted, 0,
403 "compaction must wait for the never-pulled device"
404 );
405 let present: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM sync_log WHERE seq = $1")
406 .bind(seq)
407 .fetch_one(&h.db)
408 .await
409 .unwrap();
410 assert_eq!(
411 present, 1,
412 "log entry must survive while a new device sits at seq 0"
413 );
414
415 // Once device B catches up, the same entry is safely compactable.
416 sqlx::query("UPDATE sync_devices SET last_pulled_seq = $1 WHERE device_name = 'B'")
417 .bind(seq)
418 .execute(&h.db)
419 .await
420 .unwrap();
421 let deleted = makenotwork::db::synckit::compact_all_sync_logs(&h.db, 7)
422 .await
423 .unwrap();
424 assert_eq!(
425 deleted, 1,
426 "after every device caught up, the old entry compacts"
427 );
428 }
429
430 #[tokio::test]
431 async fn delete_absent_blob_is_idempotent_no_op() {
432 let (mut h, _blobs) = harness_with_blobs().await;
433 let user_id = h
434 .signup("paid_del2", "paid_del2@example.com", "Password1!")
435 .await;
436 let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
437 seed_subscription(&h.db, user_id, app_id, "active", GIB).await;
438 auth_as(&mut h, user_id, app_id, "user-key");
439
440 // Deleting a hash that was never stored is a 204 no-op, and enqueues nothing.
441 let resp = h
442 .client
443 .delete(&format!("/api/sync/blobs/{}", fake_hash(0x08)))
444 .await;
445 assert_eq!(
446 resp.status, 204,
447 "deleting an absent blob should be a 204 no-op: {}",
448 resp.text
449 );
450
451 let enqueued: i64 = sqlx::query_scalar(
452 "SELECT COUNT(*) FROM pending_s3_deletions WHERE source = 'synckit_blob_delete'",
453 )
454 .fetch_one(&h.db)
455 .await
456 .unwrap();
457 assert_eq!(enqueued, 0, "no-op delete must not dead-letter anything");
458 }
459