diff --git a/packages/super-sync-server/src/sync/services/snapshot.service.ts b/packages/super-sync-server/src/sync/services/snapshot.service.ts index 9a984b22ad..48f21c5c72 100644 --- a/packages/super-sync-server/src/sync/services/snapshot.service.ts +++ b/packages/super-sync-server/src/sync/services/snapshot.service.ts @@ -445,9 +445,15 @@ export class SnapshotService { } } + // Always validate that the cached base is not poisoned by encrypted + // ops before serving it, including on the fast path — a legacy build + // that didn't reject encrypted ops could have produced a poisoned + // cache, and the fast path would serve it forever. The assertion is + // a findFirst + count, which is cheap relative to skipping replay. + await this._assertCachedSnapshotBaseReplayable(tx, userId, startSeq); + // Fast path: cached snapshot is already at the latest seq AND in the - // current schema version. No replay needed → skip the encrypted-op - // probes (they cost a findFirst + count per call). + // current schema version. No replay needed. if ( startSeq >= latestSeq && cachedRow?.snapshotData && @@ -461,8 +467,6 @@ export class SnapshotService { }; } - await this._assertCachedSnapshotBaseReplayable(tx, userId, startSeq); - // Migrate snapshot if needed if (stateNeedsMigration(snapshotSchemaVersion, CURRENT_SCHEMA_VERSION)) { Logger.info( @@ -534,30 +538,46 @@ export class SnapshotService { const generatedAt = Date.now(); // Cache the generated snapshot inline so it matches the returned state. - // Upsert: a user that just had a snapshot generated may still have no - // userSyncState row (first-time path), in which case `update` throws - // RecordNotFound. Capture the byte delta so the storage usage counter - // can be updated AFTER the txn commits (see `cacheDelta`). + // Race-safe write (same pattern as `cacheSnapshot`): a concurrent + // upload may have advanced `lastSnapshotSeq` beyond our `latestSeq` + // while we replayed under RepeatableRead. Conditional `updateMany` + // only writes when our seq is newer; if no row exists yet (first-time + // user), fall back to `create` and swallow P2002 if another writer + // beat us to the insert. When the race is lost, `cacheDelta` MUST + // stay 0 — otherwise `onCacheDelta` would over-credit storage by + // `compressed.length - previousCachedBytes` even though nothing was + // written. const previousCachedBytes = cachedRow?.snapshotData?.length ?? 0; const compressed = await gzipAsync(Buffer.from(JSON.stringify(state), 'utf-8')); if (compressed.length <= MAX_SNAPSHOT_SIZE_BYTES) { - await tx.userSyncState.upsert({ - where: { userId }, - update: { - snapshotData: compressed, - lastSnapshotSeq: latestSeq, - snapshotAt: BigInt(generatedAt), - snapshotSchemaVersion: CURRENT_SCHEMA_VERSION, - }, - create: { + const cacheData = { + snapshotData: compressed, + lastSnapshotSeq: latestSeq, + snapshotAt: BigInt(generatedAt), + snapshotSchemaVersion: CURRENT_SCHEMA_VERSION, + }; + const updateResult = await tx.userSyncState.updateMany({ + where: { userId, - snapshotData: compressed, - lastSnapshotSeq: latestSeq, - snapshotAt: BigInt(generatedAt), - snapshotSchemaVersion: CURRENT_SCHEMA_VERSION, + OR: [{ lastSnapshotSeq: null }, { lastSnapshotSeq: { lt: latestSeq } }], }, + data: cacheData, }); - cacheDelta = compressed.length - previousCachedBytes; + if (updateResult.count > 0) { + cacheDelta = compressed.length - previousCachedBytes; + } else { + // Either no row exists yet (first-time-user path) or a newer + // snapshot already won the race. Try a create — if a row was + // inserted concurrently, the unique-userId constraint throws + // P2002 and we treat that as "newer snapshot won". + try { + await tx.userSyncState.create({ data: { userId, ...cacheData } }); + cacheDelta = compressed.length - previousCachedBytes; + } catch (err) { + if ((err as { code?: string }).code !== 'P2002') throw err; + // cacheDelta stays 0: nothing was written. + } + } } return { @@ -896,7 +916,19 @@ export class SnapshotService { for (const key of Object.keys(state)) { delete state[key]; } - Object.assign(state, fullState); + // Copy key-by-key (not Object.assign) so a malicious `__proto__` + // key in the client-uploaded payload cannot pollute Object's + // prototype via the `__proto__` setter. JSON.parse creates + // `__proto__` as an own data property (no setter), but + // Object.assign would then `state['__proto__'] = …`, which DOES + // trigger the setter and pollute the prototype chain. + const fullStateRecord = fullState as Record; + for (const key of Object.keys(fullStateRecord)) { + if (key === '__proto__' || key === 'constructor' || key === 'prototype') { + continue; + } + state[key] = fullStateRecord[key]; + } continue; } diff --git a/packages/super-sync-server/tests/snapshot.service.spec.ts b/packages/super-sync-server/tests/snapshot.service.spec.ts index a215bf0c6b..42295e61da 100644 --- a/packages/super-sync-server/tests/snapshot.service.spec.ts +++ b/packages/super-sync-server/tests/snapshot.service.spec.ts @@ -347,10 +347,11 @@ describe('SnapshotService', () => { // Regression test for the RecordNotFound bug: when a user calls // generateSnapshot before any uploads, there is no userSyncState row. // Previously the inline `tx.userSyncState.update` threw P2025 and the - // route returned 500. With upsert the path must succeed and return an - // empty snapshot at seq=0. + // route returned 500. With the race-safe updateMany+create fallback the + // path must succeed and return an empty snapshot at seq=0. it('should not throw RecordNotFound when no userSyncState row exists', async () => { - const upsertSpy = vi.fn().mockResolvedValue({}); + const updateManySpy = vi.fn().mockResolvedValue({ count: 0 }); + const createSpy = vi.fn().mockResolvedValue({}); vi.mocked(prisma.$transaction).mockImplementation(async (fn) => { const mockTx = { @@ -358,7 +359,8 @@ describe('SnapshotService', () => { // Both findUnique calls (lastSeq + cachedRow) return null — // simulating a brand-new user. findUnique: vi.fn().mockResolvedValue(null), - upsert: upsertSpy, + updateMany: updateManySpy, + create: createSpy, }, operation: { count: vi.fn().mockResolvedValue(0), @@ -372,15 +374,13 @@ describe('SnapshotService', () => { const result = await service.generateSnapshot(1); // latestSeq=0, startSeq=0 → no replay needed but no cache either, so - // the service runs through replay (empty loop) and still upserts. + // the service runs through replay (empty loop) and still writes the + // cache row via updateMany+create fallback. expect(result.serverSeq).toBe(0); expect(result.state).toEqual({}); - expect(upsertSpy).toHaveBeenCalledWith( - expect.objectContaining({ - where: { userId: 1 }, - create: expect.objectContaining({ userId: 1, lastSnapshotSeq: 0 }), - }), - ); + expect(createSpy).toHaveBeenCalledWith({ + data: expect.objectContaining({ userId: 1, lastSnapshotSeq: 0 }), + }); }); it('cacheSnapshot via the standalone path should create the row on first-time user', async () => { @@ -423,7 +423,8 @@ describe('SnapshotService', () => { snapshotAt: null, snapshotSchemaVersion: null, }), - upsert: vi.fn().mockResolvedValue({}), + updateMany: vi.fn().mockResolvedValue({ count: 0 }), + create: vi.fn().mockResolvedValue({}), }, operation: { count: vi.fn().mockResolvedValue(0), findMany }, }; @@ -453,7 +454,8 @@ describe('SnapshotService', () => { // MAX_SNAPSHOT_SIZE_BYTES with no quota accounting. const previousBytes = 500; vi.mocked(prisma.$transaction).mockImplementation(async (fn: any) => { - const upsert = vi.fn().mockResolvedValue({}); + const updateMany = vi.fn().mockResolvedValue({ count: 1 }); + const create = vi.fn().mockResolvedValue({}); const mockTx = { userSyncState: { findUnique: vi @@ -465,7 +467,8 @@ describe('SnapshotService', () => { snapshotAt: null, snapshotSchemaVersion: 1, }), - upsert, + updateMany, + create, }, operation: { count: vi.fn().mockResolvedValue(0), @@ -504,8 +507,14 @@ describe('SnapshotService', () => { it('should not invoke onCacheDelta when the cached snapshot is already up to date', async () => { // When startSeq >= latestSeq the snapshot service returns the cached // state without rewriting snapshotData — no counter update is needed. + // The fast path now runs _assertCachedSnapshotBaseReplayable first, so + // the operation mocks must answer the findFirst/count probes. Set + // snapshotSchemaVersion to CURRENT_SCHEMA_VERSION (2) so the fast + // path triggers; v1 would force a migration and bypass it. const cachedState = { TASK: { t1: { id: 't1' } } }; const compressed = zlib.gzipSync(JSON.stringify(cachedState)); + const updateManySpy = vi.fn().mockResolvedValue({ count: 0 }); + const createSpy = vi.fn(); vi.mocked(prisma.$transaction).mockImplementation(async (fn: any) => { const mockTx = { userSyncState: { @@ -516,11 +525,16 @@ describe('SnapshotService', () => { snapshotData: compressed, lastSnapshotSeq: 7, snapshotAt: BigInt(1), - snapshotSchemaVersion: 1, + snapshotSchemaVersion: 2, }), - upsert: vi.fn(), + updateMany: updateManySpy, + create: createSpy, + }, + operation: { + count: vi.fn().mockResolvedValue(0), + findFirst: vi.fn().mockResolvedValue({ serverSeq: 7 }), + findMany: vi.fn(), }, - operation: { count: vi.fn(), findFirst: vi.fn(), findMany: vi.fn() }, }; return fn(mockTx); }); @@ -530,6 +544,50 @@ describe('SnapshotService', () => { await service.generateSnapshot(1, onCacheDelta); expect(onCacheDelta).not.toHaveBeenCalled(); + // Fast path returns before the cache-write block, so neither write op + // is invoked. + expect(updateManySpy).not.toHaveBeenCalled(); + expect(createSpy).not.toHaveBeenCalled(); + }); + + it('should run _assertCachedSnapshotBaseReplayable before the fast path returns the cached snapshot', async () => { + // Defense in depth: a legacy server build that failed to reject + // encrypted ops could have produced a poisoned cache. The fast path + // (startSeq >= latestSeq + schema match) MUST still validate the + // cached base, otherwise it would serve poisoned cache forever. + const cachedState = { TASK: { t1: { id: 't1' } } }; + const compressed = zlib.gzipSync(JSON.stringify(cachedState)); + const findFirstSpy = vi.fn().mockResolvedValue(null); + const countSpy = vi.fn().mockResolvedValue(1); // encrypted op present + vi.mocked(prisma.$transaction).mockImplementation(async (fn: any) => { + const mockTx = { + userSyncState: { + findUnique: vi + .fn() + .mockResolvedValueOnce({ lastSeq: 7 }) + .mockResolvedValueOnce({ + snapshotData: compressed, + lastSnapshotSeq: 7, + snapshotAt: BigInt(1), + snapshotSchemaVersion: 2, + }), + updateMany: vi.fn(), + create: vi.fn(), + }, + operation: { + findFirst: findFirstSpy, + count: countSpy, + findMany: vi.fn(), + }, + }; + return fn(mockTx); + }); + + await expect(service.generateSnapshot(1)).rejects.toThrow( + 'ENCRYPTED_OPS_NOT_SUPPORTED', + ); + expect(findFirstSpy).toHaveBeenCalled(); + expect(countSpy).toHaveBeenCalled(); }); it('should swallow a thrown onCacheDelta to avoid corrupting the snapshot result', async () => { @@ -547,7 +605,8 @@ describe('SnapshotService', () => { snapshotAt: null, snapshotSchemaVersion: null, }), - upsert: vi.fn().mockResolvedValue({}), + updateMany: vi.fn().mockResolvedValue({ count: 0 }), + create: vi.fn().mockResolvedValue({}), }, operation: { count: vi.fn().mockResolvedValue(0), @@ -700,7 +759,8 @@ describe('SnapshotService', () => { const mockTx = { userSyncState: { findUnique: vi.fn().mockResolvedValue({ lastSeq: 0 }), - upsert: vi.fn().mockResolvedValue({}), + updateMany: vi.fn().mockResolvedValue({ count: 0 }), + create: vi.fn().mockResolvedValue({}), }, operation: { findMany: vi.fn().mockResolvedValue([]), @@ -733,7 +793,8 @@ describe('SnapshotService', () => { const mockTx = { userSyncState: { findUnique: vi.fn().mockResolvedValue({ lastSeq: 0 }), - upsert: vi.fn().mockResolvedValue({}), + updateMany: vi.fn().mockResolvedValue({ count: 0 }), + create: vi.fn().mockResolvedValue({}), }, operation: { findMany: vi.fn().mockResolvedValue([]), @@ -790,7 +851,8 @@ describe('SnapshotService', () => { .fn() .mockResolvedValueOnce({ lastSeq: 1 }) .mockResolvedValueOnce({ snapshotData: null, lastSnapshotSeq: null }), - upsert: vi.fn().mockResolvedValue({}), + updateMany: vi.fn().mockResolvedValue({ count: 0 }), + create: vi.fn().mockResolvedValue({}), }, operation: { count: vi.fn().mockResolvedValue(0), @@ -849,7 +911,8 @@ describe('SnapshotService', () => { // Scenario: after a clean-slate upload (sync.service preserves lastSeq but // wipes ops) the next SYNC_IMPORT lands at lastSeq+1 (e.g. 102). Snapshot // replay must accept the leading gap because SYNC_IMPORT resets state. - const upsertSpy = vi.fn().mockResolvedValue({}); + const updateManySpy = vi.fn().mockResolvedValue({ count: 0 }); + const createSpy = vi.fn().mockResolvedValue({}); vi.mocked(prisma.$transaction).mockImplementation(async (fn) => { const mockTx = { @@ -858,7 +921,8 @@ describe('SnapshotService', () => { .fn() .mockResolvedValueOnce({ lastSeq: 102 }) .mockResolvedValueOnce({ snapshotData: null, lastSnapshotSeq: null }), - upsert: upsertSpy, + updateMany: updateManySpy, + create: createSpy, }, operation: { count: vi.fn().mockResolvedValue(0), @@ -945,7 +1009,8 @@ describe('SnapshotService', () => { const mockTx = { userSyncState: { findUnique: vi.fn().mockResolvedValue({ lastSeq: 0 }), - upsert: vi.fn().mockResolvedValue({}), + updateMany: vi.fn().mockResolvedValue({ count: 0 }), + create: vi.fn().mockResolvedValue({}), }, operation: { findMany: vi.fn().mockResolvedValue([]), @@ -1461,6 +1526,54 @@ describe('SnapshotService', () => { expect(result.TAG).toBeUndefined(); }); + it('SYNC_IMPORT must not allow prototype pollution via __proto__ in the uploaded payload', () => { + // Regression: a malicious client could upload a SYNC_IMPORT payload + // whose `appDataComplete` contains a `__proto__` key. The previous + // implementation used Object.assign(state, fullState) which triggers + // the prototype setter (Object.assign uses [[Set]] semantics) and + // pollutes Object.prototype for every plain object in the process — + // a server-wide compromise from a single malicious upload. + // + // Construct a malicious payload via JSON.parse so __proto__ is an + // own property (V8 / spec behaviour since 2018). + const maliciousPayload = JSON.parse( + JSON.stringify({ + appDataComplete: { + TASK: { 'new-task': { id: 'new-task' } }, + }, + }).replace('"TASK"', '"__proto__":{"polluted":true},"TASK"'), + ); + // Sanity-check the test fixture itself before relying on it. + const inner = maliciousPayload.appDataComplete as Record; + expect(Object.prototype.hasOwnProperty.call(inner, '__proto__')).toBe(true); + + const ops = [ + { + id: 'op-malicious', + opType: 'SYNC_IMPORT', + entityType: 'FULL_STATE', + entityId: null, + payload: maliciousPayload, + isPayloadEncrypted: false, + serverSeq: 1, + schemaVersion: 1, + }, + ]; + + try { + service.replayOpsToState(ops as any); + // The benign keys must be copied. + // (Object.prototype itself must not gain a `polluted` property.) + // Cast through unknown because TS narrows {} to never for arbitrary + // index access. + expect(({} as { polluted?: unknown }).polluted).toBeUndefined(); + expect((Object.prototype as { polluted?: unknown }).polluted).toBeUndefined(); + } finally { + // Defensive cleanup in case the assertion fires after pollution. + delete (Object.prototype as { polluted?: unknown }).polluted; + } + }); + it('should throw EncryptedOpsNotSupportedError for encrypted operations', () => { const ops = [ { diff --git a/packages/super-sync-server/tests/sync-operations.spec.ts b/packages/super-sync-server/tests/sync-operations.spec.ts index 3ba2ce37ef..5e98b4db8a 100644 --- a/packages/super-sync-server/tests/sync-operations.spec.ts +++ b/packages/super-sync-server/tests/sync-operations.spec.ts @@ -219,6 +219,33 @@ vi.mock('../src/db', async () => { state.userSyncStates.set(args.where.userId, result); return result; }), + // Race-safe writers used by SnapshotService._generateSnapshotImpl: + // updateMany honours the `OR: [lastSnapshotSeq null, lastSnapshotSeq < seq]` + // guard; create is the first-time-user fallback. + updateMany: vi.fn().mockImplementation(async (args: any) => { + const existing = state.userSyncStates.get(args.where.userId); + if (!existing) return { count: 0 }; + const seqFilter = args.where?.OR?.find( + (clause: any) => clause.lastSnapshotSeq?.lt !== undefined, + )?.lastSnapshotSeq?.lt; + const cachedSeq = existing.lastSnapshotSeq; + const matches = + cachedSeq === null || + cachedSeq === undefined || + (seqFilter !== undefined && cachedSeq < seqFilter); + if (!matches) return { count: 0 }; + const updated = { ...existing, ...args.data }; + state.userSyncStates.set(args.where.userId, updated); + return { count: 1 }; + }), + create: vi.fn().mockImplementation(async (args: any) => { + if (state.userSyncStates.has(args.data.userId)) { + throw Object.assign(new Error('Unique constraint'), { code: 'P2002' }); + } + const result = { ...args.data }; + state.userSyncStates.set(args.data.userId, result); + return result; + }), update: vi.fn().mockImplementation(async (args: any) => { const existing = state.userSyncStates.get(args.where.userId); if (existing) { diff --git a/packages/super-sync-server/tests/time-tracking-operations.spec.ts b/packages/super-sync-server/tests/time-tracking-operations.spec.ts index 2dd8f341f3..dd7e3deeaf 100644 --- a/packages/super-sync-server/tests/time-tracking-operations.spec.ts +++ b/packages/super-sync-server/tests/time-tracking-operations.spec.ts @@ -218,6 +218,33 @@ vi.mock('../src/db', async () => { state.userSyncStates.set(args.where.userId, result); return result; }), + // Race-safe writers used by SnapshotService._generateSnapshotImpl: + // updateMany honours the `OR: [lastSnapshotSeq null, lastSnapshotSeq < seq]` + // guard; create is the first-time-user fallback. + updateMany: vi.fn().mockImplementation(async (args: any) => { + const existing = state.userSyncStates.get(args.where.userId); + if (!existing) return { count: 0 }; + const seqFilter = args.where?.OR?.find( + (clause: any) => clause.lastSnapshotSeq?.lt !== undefined, + )?.lastSnapshotSeq?.lt; + const cachedSeq = existing.lastSnapshotSeq; + const matches = + cachedSeq === null || + cachedSeq === undefined || + (seqFilter !== undefined && cachedSeq < seqFilter); + if (!matches) return { count: 0 }; + const updated = { ...existing, ...args.data }; + state.userSyncStates.set(args.where.userId, updated); + return { count: 1 }; + }), + create: vi.fn().mockImplementation(async (args: any) => { + if (state.userSyncStates.has(args.data.userId)) { + throw Object.assign(new Error('Unique constraint'), { code: 'P2002' }); + } + const result = { ...args.data }; + state.userSyncStates.set(args.data.userId, result); + return result; + }), update: vi.fn().mockImplementation(async (args: any) => { const existing = state.userSyncStates.get(args.where.userId); if (existing) {