diff --git a/cloudflare/src/api_controls.ts b/cloudflare/src/api_controls.ts index 750d1b8..fbe970b 100644 --- a/cloudflare/src/api_controls.ts +++ b/cloudflare/src/api_controls.ts @@ -13,6 +13,12 @@ const RATE_LIMIT_WINDOW_SECONDS = 60; export type ApiHandler = () => Promise; export type AuthenticatedApiHandler = (context: AuthContext) => Promise; +const APPROVED_DEVICE_QUERY = ` + SELECT device_id + FROM user_devices + WHERE user_id = ? AND device_id = ? AND approval_status = 'approved' AND revoked_at IS NULL +`; + export async function withPublicApiControls( request: Request, env: Env, @@ -103,6 +109,37 @@ export async function withAuthenticatedApiControls( } } +export async function withApprovedDeviceApiControls( + request: Request, + env: Env, + route: string, + allowedMethods: readonly string[], + handler: AuthenticatedApiHandler, +): Promise { + return withAuthenticatedApiControls(request, env, route, allowedMethods, async (context) => { + if (context.deviceId === undefined) { + return jsonResponse({ error: "device_context_required" }, 403, { + "Cache-Control": "no-store", + }); + } + + const row = await env.ELY_DB.prepare(APPROVED_DEVICE_QUERY) + .bind(context.userId, context.deviceId) + .first(); + if (row === null) { + return jsonResponse({ error: "device_not_approved" }, 403, { + "Cache-Control": "no-store", + }); + } + + return handler(context); + }); +} + +interface ApprovedDeviceRow { + device_id: unknown; +} + function rateLimitKey(environment: string, route: string): string { return `${environment}:${route}`; } diff --git a/cloudflare/src/index.ts b/cloudflare/src/index.ts index 3f2bfb3..29858b0 100644 --- a/cloudflare/src/index.ts +++ b/cloudflare/src/index.ts @@ -1,5 +1,9 @@ import type { Env } from "./bindings.js"; -import { withAuthenticatedApiControls, withPublicApiControls } from "./api_controls.js"; +import { + withApprovedDeviceApiControls, + withAuthenticatedApiControls, + withPublicApiControls, +} from "./api_controls.js"; import { DevicePermissionError, DevicePersistenceError, @@ -31,6 +35,7 @@ import { parsePublicSigningKeysDocument, publicSigningKeysKvKey, } from "./signing_keys.js"; +import { SyncRequestError, SyncSchemaError, syncPullDocument } from "./sync_pull.js"; import { jsonResponse } from "./responses.js"; export default { @@ -169,6 +174,37 @@ export async function handleRequest(request: Request, env: Env): Promise { + try { + return jsonResponse(await syncPullDocument(url, env, context), 200, { + "Cache-Control": "no-store", + }); + } catch (error) { + if (error instanceof SyncRequestError) { + return jsonResponse( + { error: "invalid_sync_pull" }, + 400, + { "Cache-Control": "no-store" }, + ); + } + if (error instanceof SyncSchemaError) { + return jsonResponse( + { error: "sync_pull_invalid" }, + 500, + { "Cache-Control": "no-store" }, + ); + } + throw error; + } + }, + ); + } if (url.pathname === "/api/plugins/signing-keys") { return withPublicApiControls(request, env, "plugins.signing_keys", ["GET"], () => handlePublicSigningKeys(env), diff --git a/cloudflare/src/sync_pull.ts b/cloudflare/src/sync_pull.ts new file mode 100644 index 0000000..b09f46a --- /dev/null +++ b/cloudflare/src/sync_pull.ts @@ -0,0 +1,201 @@ +import type { AuthContext } from "./auth.js"; +import type { Env } from "./bindings.js"; + +const DEFAULT_SYNC_PULL_LIMIT = 100; +const MAX_SYNC_PULL_LIMIT = 500; +const SYNC_OBJECT_ID_PATTERN = /^[a-zA-Z0-9._:-]{1,128}$/; +const SYNC_OBJECT_TYPE_PATTERN = /^[a-z0-9][a-z0-9._:-]{0,127}$/; +const SHA256_HEX = /^[a-f0-9]{64}$/; + +const SYNC_CHANGE_LOG_QUERY = ` + SELECT + change_id, + object_id, + object_type, + operation, + payload_hash, + logical_clock, + device_id, + created_at + FROM sync_change_log + WHERE user_id = ? AND change_id > ? + ORDER BY change_id ASC + LIMIT ? +`; + +export interface SyncPullDocument { + version: 1; + user_id: string; + device_id: string; + cursor: number; + next_cursor: number; + has_more: boolean; + changes: SyncChangeDocument[]; +} + +export interface SyncChangeDocument { + change_id: number; + object_id: string; + object_type: string; + operation: "upsert" | "delete"; + payload_hash: string; + logical_clock: number; + device_id: string; + created_at: number; +} + +interface SyncChangeRow { + change_id: unknown; + object_id: unknown; + object_type: unknown; + operation: unknown; + payload_hash: unknown; + logical_clock: unknown; + device_id: unknown; + created_at: unknown; +} + +interface SyncPullQuery { + cursor: number; + limit: number; +} + +export class SyncSchemaError extends Error { + constructor(message: string) { + super(message); + this.name = "SyncSchemaError"; + } +} + +export class SyncRequestError extends Error { + constructor(message: string) { + super(message); + this.name = "SyncRequestError"; + } +} + +export async function syncPullDocument( + url: URL, + env: Env, + context: AuthContext, +): Promise { + const deviceId = currentDeviceId(context); + const query = syncPullQuery(url); + const result = await env.ELY_DB.prepare(SYNC_CHANGE_LOG_QUERY) + .bind(context.userId, query.cursor, query.limit + 1) + .all(); + const rows = result.results.slice(0, query.limit); + const changes = rows.map(syncChangeDocument); + return { + version: 1, + user_id: context.userId, + device_id: deviceId, + cursor: query.cursor, + next_cursor: changes.at(-1)?.change_id ?? query.cursor, + has_more: result.results.length > query.limit, + changes, + }; +} + +function syncPullQuery(url: URL): SyncPullQuery { + assertOnlyQueryParams(url, ["cursor", "limit"]); + const cursor = requiredQueryInteger(url, "cursor", 0, Number.MAX_SAFE_INTEGER); + const limit = optionalQueryInteger(url, "limit", 1, MAX_SYNC_PULL_LIMIT) ?? DEFAULT_SYNC_PULL_LIMIT; + return { cursor, limit }; +} + +function syncChangeDocument(row: SyncChangeRow): SyncChangeDocument { + return { + change_id: integerValue(row.change_id, "change_id", 0, Number.MAX_SAFE_INTEGER), + object_id: objectId(row.object_id), + object_type: objectType(row.object_type), + operation: operation(row.operation), + payload_hash: payloadHash(row.payload_hash), + logical_clock: integerValue(row.logical_clock, "logical_clock", 0, Number.MAX_SAFE_INTEGER), + device_id: objectId(row.device_id), + created_at: integerValue(row.created_at, "created_at", 0, Number.MAX_SAFE_INTEGER), + }; +} + +function currentDeviceId(context: AuthContext): string { + if (context.deviceId === undefined) { + throw new SyncSchemaError("device_context_required"); + } + return context.deviceId; +} + +function assertOnlyQueryParams(url: URL, fields: string[]): void { + const allowed = new Set(fields); + for (const field of url.searchParams.keys()) { + if (!allowed.has(field)) { + throw new SyncRequestError(`unexpected_query:${field}`); + } + } +} + +function requiredQueryInteger(url: URL, field: string, min: number, max: number): number { + const value = url.searchParams.get(field); + if (value === null) { + throw new SyncRequestError(`${field}_required`); + } + return queryInteger(value, field, min, max); +} + +function optionalQueryInteger( + url: URL, + field: string, + min: number, + max: number, +): number | undefined { + const value = url.searchParams.get(field); + if (value === null) { + return undefined; + } + return queryInteger(value, field, min, max); +} + +function queryInteger(value: string, field: string, min: number, max: number): number { + if (!/^[0-9]+$/.test(value)) { + throw new SyncRequestError(`${field}_invalid`); + } + const parsed = Number(value); + if (!Number.isSafeInteger(parsed) || parsed < min || parsed > max) { + throw new SyncRequestError(`${field}_invalid`); + } + return parsed; +} + +function integerValue(value: unknown, field: string, min: number, max: number): number { + if (typeof value !== "number" || !Number.isSafeInteger(value) || value < min || value > max) { + throw new SyncSchemaError(`${field}_invalid`); + } + return value; +} + +function objectId(value: unknown): string { + if (typeof value !== "string" || !SYNC_OBJECT_ID_PATTERN.test(value)) { + throw new SyncSchemaError("object_id_invalid"); + } + return value; +} + +function objectType(value: unknown): string { + if (typeof value !== "string" || !SYNC_OBJECT_TYPE_PATTERN.test(value)) { + throw new SyncSchemaError("object_type_invalid"); + } + return value; +} + +function operation(value: unknown): SyncChangeDocument["operation"] { + if (value !== "upsert" && value !== "delete") { + throw new SyncSchemaError("operation_invalid"); + } + return value; +} + +function payloadHash(value: unknown): string { + if (typeof value !== "string" || !SHA256_HEX.test(value)) { + throw new SyncSchemaError("payload_hash_invalid"); + } + return value; +} diff --git a/cloudflare/tests/sync_pull_routes.test.ts b/cloudflare/tests/sync_pull_routes.test.ts new file mode 100644 index 0000000..1f975a3 --- /dev/null +++ b/cloudflare/tests/sync_pull_routes.test.ts @@ -0,0 +1,169 @@ +import assert from "node:assert/strict"; +import { describe, it } from "node:test"; + +import { authSessionCacheKvKey, authTokenHash } from "../src/auth.js"; +import { handleRequest } from "../src/index.js"; +import { ACCESS_TOKEN, sessionDocument, testD1Database, testEnv } from "./devices_test_support.js"; + +const PAYLOAD_HASH = "a".repeat(64); + +describe("sync pull routes", () => { + it("returns sync change log entries for an approved current device", async () => { + const tokenHash = await authTokenHash(ACCESS_TOKEN); + const d1 = testD1Database({ + firstRows: [{ device_id: "device-01" }], + allRows: [ + syncChangeRow({ change_id: 11, object_id: "tab-01" }), + syncChangeRow({ change_id: 12, object_id: "bookmark-01", object_type: "bookmarks" }), + ], + }); + + const response = await handleRequest( + new Request("https://elydora.test/api/sync/pull?cursor=10&limit=2", { + headers: { authorization: `Bearer ${ACCESS_TOKEN}` }, + }), + testEnv({ + d1, + kvEntries: [[authSessionCacheKvKey("local", tokenHash), sessionDocument("device-01")]], + }), + ); + + assert.equal(response.status, 200); + assert.equal(response.headers.get("cache-control"), "no-store"); + assert.deepEqual(await response.json(), { + version: 1, + user_id: "user-01", + device_id: "device-01", + cursor: 10, + next_cursor: 12, + has_more: false, + changes: [ + syncChangeDocument({ change_id: 11, object_id: "tab-01" }), + syncChangeDocument({ change_id: 12, object_id: "bookmark-01", object_type: "bookmarks" }), + ], + }); + assert.ok(d1.queries[0]?.includes("approval_status = 'approved'")); + assert.ok(d1.queries[1]?.includes("FROM sync_change_log")); + assert.deepEqual(d1.binds, [ + ["user-01", "device-01"], + ["user-01", 10, 3], + ]); + }); + + it("reports more changes when the pull window is saturated", async () => { + const tokenHash = await authTokenHash(ACCESS_TOKEN); + const d1 = testD1Database({ + firstRows: [{ device_id: "device-01" }], + allRows: [ + syncChangeRow({ change_id: 11, object_id: "tab-01" }), + syncChangeRow({ change_id: 12, object_id: "tab-02" }), + syncChangeRow({ change_id: 13, object_id: "tab-03" }), + ], + }); + + const response = await handleRequest( + new Request("https://elydora.test/api/sync/pull?cursor=10&limit=2", { + headers: { authorization: `Bearer ${ACCESS_TOKEN}` }, + }), + testEnv({ + d1, + kvEntries: [[authSessionCacheKvKey("local", tokenHash), sessionDocument("device-01")]], + }), + ); + + const body = (await response.json()) as { has_more: boolean; next_cursor: number; changes: [] }; + assert.equal(response.status, 200); + assert.equal(body.has_more, true); + assert.equal(body.next_cursor, 12); + assert.equal(body.changes.length, 2); + }); + + it("rejects revoked devices before reading sync deltas", async () => { + const tokenHash = await authTokenHash(ACCESS_TOKEN); + const d1 = testD1Database({ firstRows: [null], allRows: [syncChangeRow()] }); + + const response = await handleRequest( + new Request("https://elydora.test/api/sync/pull?cursor=10", { + headers: { authorization: `Bearer ${ACCESS_TOKEN}` }, + }), + testEnv({ + d1, + kvEntries: [[authSessionCacheKvKey("local", tokenHash), sessionDocument("device-01")]], + }), + ); + + assert.equal(response.status, 403); + assert.deepEqual(await response.json(), { error: "device_not_approved" }); + assert.equal(d1.queries.length, 1); + }); + + it("rejects invalid cursors after session and device validation", async () => { + const tokenHash = await authTokenHash(ACCESS_TOKEN); + const d1 = testD1Database({ firstRows: [{ device_id: "device-01" }] }); + + const response = await handleRequest( + new Request("https://elydora.test/api/sync/pull?cursor=old", { + headers: { authorization: `Bearer ${ACCESS_TOKEN}` }, + }), + testEnv({ + d1, + kvEntries: [[authSessionCacheKvKey("local", tokenHash), sessionDocument("device-01")]], + }), + ); + + assert.equal(response.status, 400); + assert.deepEqual(await response.json(), { error: "invalid_sync_pull" }); + assert.equal(d1.queries.length, 1); + }); + + it("returns a server error for malformed sync change rows", async () => { + const tokenHash = await authTokenHash(ACCESS_TOKEN); + const d1 = testD1Database({ + firstRows: [{ device_id: "device-01" }], + allRows: [syncChangeRow({ payload_hash: "bad" })], + }); + + const response = await handleRequest( + new Request("https://elydora.test/api/sync/pull?cursor=10", { + headers: { authorization: `Bearer ${ACCESS_TOKEN}` }, + }), + testEnv({ + d1, + kvEntries: [[authSessionCacheKvKey("local", tokenHash), sessionDocument("device-01")]], + }), + ); + + assert.equal(response.status, 500); + assert.deepEqual(await response.json(), { error: "sync_pull_invalid" }); + }); + + it("rejects unauthenticated sync pulls before D1 reads", async () => { + const d1 = testD1Database({ allRows: [syncChangeRow()] }); + const response = await handleRequest( + new Request("https://elydora.test/api/sync/pull?cursor=0"), + testEnv({ d1 }), + ); + + assert.equal(response.status, 401); + assert.deepEqual(await response.json(), { error: "authorization_missing" }); + assert.deepEqual(d1.queries, []); + }); +}); + +function syncChangeRow(overrides: Record = {}): Record { + return { + change_id: 11, + object_id: "tab-01", + object_type: "tabs", + operation: "upsert", + payload_hash: PAYLOAD_HASH, + logical_clock: 42, + device_id: "device-02", + created_at: 1_780_000_500, + ...overrides, + }; +} + +function syncChangeDocument(overrides: Record = {}): Record { + return syncChangeRow(overrides); +}