From 27248af69a046df4b0d8ec44d77fe63b87dd19b9 Mon Sep 17 00:00:00 2001 From: xiduzo Date: Mon, 3 Aug 2026 09:02:00 +0200 Subject: [PATCH] fix(collab): authorize the user for the flow on the Yjs websocket The /yjs/:flowId endpoint authenticated the caller but never authorized them for that flowId: getOrCreateRoom loaded the document by id alone and persistRoom wrote it back by id alone, so any authenticated user could both read and write any flow's live document, bypassing the owner and collaborator model the tRPC procedures enforce via requireFlowAccess. Authorize at the websocket boundary, before the room is created, reusing requireFlowAccess so there stays one source of truth for who counts as what. The resolved role also decides write access: viewers were only held read-only on the client, and Yjs is bidirectional, so an invited viewer could still write through the socket. handleConnection now takes canWrite as a required argument (no default, so no caller grants write by omission), and read-only connections may only send sync step 1 -- step 2 and update both write into the doc. Viewers still read: the client sends its own step 1 on connect and the server answers with step 2. Reported by Eesh Saxena (github.com/eeshsaxena) through SECURITY.md. Co-authored-by: eeshsaxena <139802361+eeshsaxena@users.noreply.github.com> Co-Authored-By: Claude Opus 5 --- apps/server/src/index.ts | 25 ++- .../src/__tests__/yjs-server-access.test.ts | 147 ++++++++++++++++++ packages/collab/src/handler.ts | 10 +- packages/collab/src/yjs-server.ts | 19 +++ 4 files changed, 195 insertions(+), 6 deletions(-) create mode 100644 packages/collab/src/__tests__/yjs-server-access.test.ts diff --git a/apps/server/src/index.ts b/apps/server/src/index.ts index 85f0ca68..fb29088b 100644 --- a/apps/server/src/index.ts +++ b/apps/server/src/index.ts @@ -2,6 +2,7 @@ import { trpcServer } from "@hono/trpc-server"; import { createContext } from "@microflow/api/context"; import { appRouter } from "@microflow/api/routers/index"; import { getPublicSupportersCached } from "@microflow/api/routers/supporters"; +import { requireFlowAccess } from "@microflow/api/routers/flow-access"; import { auth } from "@microflow/auth"; import { env } from "@microflow/env/server"; import { createYjsHandler } from "@microflow/collab/server"; @@ -89,9 +90,27 @@ app.get( return; } - // Attach flowId and userId to the websocket - (ws.raw as unknown as { flowId: string; userId: string }).flowId = flowId; - (ws.raw as unknown as { flowId: string; userId: string }).userId = session.user.id; + // Authenticated is not authorized: check this user against THIS flow, + // through the same owner/collaborator model the tRPC procedures use. + // Viewers may connect, but read-only. + let canWrite: boolean; + try { + const { role } = await requireFlowAccess(flowId, session.user.id, "viewer"); + canWrite = role !== "viewer"; + } catch { + ws.close(1008, "Forbidden"); + return; + } + + // Attach flowId, userId and the access decision to the websocket + const data = ws.raw as unknown as { + flowId: string; + userId: string; + canWrite: boolean; + }; + data.flowId = flowId; + data.userId = session.user.id; + data.canWrite = canWrite; handler.onOpen(event, ws as any); }, diff --git a/packages/collab/src/__tests__/yjs-server-access.test.ts b/packages/collab/src/__tests__/yjs-server-access.test.ts new file mode 100644 index 00000000..003617f3 --- /dev/null +++ b/packages/collab/src/__tests__/yjs-server-access.test.ts @@ -0,0 +1,147 @@ +import { beforeEach, describe, expect, mock, test } from "bun:test"; +import * as Y from "yjs"; +import * as syncProtocol from "y-protocols/sync"; +import * as encoding from "lib0/encoding"; +import * as decoding from "lib0/decoding"; + +// The room loads/persists through the db; stub it so this test needs no +// DATABASE_URL. `ydoc: null` => the server starts from an empty document. +const updates: unknown[] = []; +mock.module("@microflow/db", () => ({ + db: { + query: { flow: { findFirst: async () => ({ id: "flow-1", ydoc: null }) } }, + update: () => ({ set: () => ({ where: async (w: unknown) => updates.push(w) }) }), + }, +})); + +const { YjsServer } = await import("../yjs-server"); + +const MESSAGE_SYNC = 0; + +/** A client's "here is my state" message — the write path. */ +function updateMessage(mutate: (doc: Y.Doc) => void): Uint8Array { + const client = new Y.Doc(); + mutate(client); + const encoder = encoding.createEncoder(); + encoding.writeVarUint(encoder, MESSAGE_SYNC); + syncProtocol.writeUpdate(encoder, Y.encodeStateAsUpdate(client)); + return encoding.toUint8Array(encoder); +} + +/** A client's "what have you got?" message — the read path. */ +function syncStep1Message(doc: Y.Doc): Uint8Array { + const encoder = encoding.createEncoder(); + encoding.writeVarUint(encoder, MESSAGE_SYNC); + syncProtocol.writeSyncStep1(encoder, doc); + return encoding.toUint8Array(encoder); +} + +function connect(server: InstanceType, canWrite: boolean) { + const received: Uint8Array[] = []; + const connection = { send: (d: Uint8Array) => received.push(d), close: () => {} }; + return { + connection, + received, + open: () => server.handleConnection("flow-1", connection, "user-1", canWrite), + }; +} + +describe("YjsServer write authorization", () => { + let server: InstanceType; + + beforeEach(() => { + server = new YjsServer({ persistDebounce: 60_000 }); + }); + + test("an editor connection's update is applied to the room", async () => { + const editor = connect(server, true); + await editor.open(); + + server.handleMessage( + "flow-1", + editor.connection, + updateMessage((doc) => doc.getMap("nodes").set("a", "editor-wrote-this"), + ), + ); + + // Reading back through a second connection's sync step 1 proves the room + // doc actually changed. + const observer = connect(server, false); + await observer.open(); + const mirror = new Y.Doc(); + for (const message of observer.received) { + const decoder = decoding.createDecoder(message); + if (decoding.readVarUint(decoder) !== MESSAGE_SYNC) continue; + syncProtocol.readSyncMessage(decoder, encoding.createEncoder(), mirror, null); + } + server.handleMessage("flow-1", observer.connection, syncStep1Message(mirror)); + for (const message of observer.received) { + const decoder = decoding.createDecoder(message); + if (decoding.readVarUint(decoder) !== MESSAGE_SYNC) continue; + syncProtocol.readSyncMessage(decoder, encoding.createEncoder(), mirror, null); + } + + expect(mirror.getMap("nodes").get("a")).toBe("editor-wrote-this"); + }); + + test("a read-only connection's update is dropped, and it can still read", async () => { + const editor = connect(server, true); + await editor.open(); + server.handleMessage( + "flow-1", + editor.connection, + updateMessage((doc) => doc.getMap("nodes").set("a", "owned")), + ); + + const viewer = connect(server, false); + await viewer.open(); + // A distinct key, so this asserts the write was dropped rather than + // relying on Y.Map's clientID-ordered conflict resolution. + server.handleMessage( + "flow-1", + viewer.connection, + updateMessage((doc) => doc.getMap("nodes").set("b", "viewer-wrote-this")), + ); + + const mirror = new Y.Doc(); + server.handleMessage("flow-1", viewer.connection, syncStep1Message(mirror)); + for (const message of viewer.received) { + const decoder = decoding.createDecoder(message); + if (decoding.readVarUint(decoder) !== MESSAGE_SYNC) continue; + syncProtocol.readSyncMessage(decoder, encoding.createEncoder(), mirror, null); + } + + // The viewer's write never lands... + expect(mirror.getMap("nodes").has("b")).toBe(false); + // ...but the viewer still receives the document (read access intact). + expect(mirror.getMap("nodes").get("a")).toBe("owned"); + }); + + test("a connection the room does not know is treated as read-only", async () => { + const editor = connect(server, true); + await editor.open(); + server.handleMessage( + "flow-1", + editor.connection, + updateMessage((doc) => doc.getMap("nodes").set("a", "owned")), + ); + + const stranger = { send: () => {}, close: () => {} }; + server.handleMessage( + "flow-1", + stranger, + updateMessage((doc) => doc.getMap("nodes").set("b", "stranger-wrote-this")), + ); + + const mirror = new Y.Doc(); + server.handleMessage("flow-1", editor.connection, syncStep1Message(mirror)); + for (const message of editor.received) { + const decoder = decoding.createDecoder(message); + if (decoding.readVarUint(decoder) !== MESSAGE_SYNC) continue; + syncProtocol.readSyncMessage(decoder, encoding.createEncoder(), mirror, null); + } + + expect(mirror.getMap("nodes").has("b")).toBe(false); + expect(mirror.getMap("nodes").get("a")).toBe("owned"); + }); +}); diff --git a/packages/collab/src/handler.ts b/packages/collab/src/handler.ts index 4a12ff3a..6f90e263 100644 --- a/packages/collab/src/handler.ts +++ b/packages/collab/src/handler.ts @@ -14,6 +14,8 @@ const yjsServer = new YjsServer(); type WebSocketData = { flowId: string; userId: string; + /** Set by the endpoint after authorizing the user on this flow. */ + canWrite: boolean; cleanup?: () => void; }; @@ -24,10 +26,11 @@ type WebSocketData = { export function createYjsHandler() { return { onOpen: async (_event: Event, ws: WSContext) => { - const { flowId, userId } = ws.raw as unknown as WebSocketData; + const { flowId, userId, canWrite } = ws.raw as unknown as WebSocketData; - if (!flowId || !userId) { - ws.close(1008, "Missing flowId or userId"); + if (!flowId || !userId || typeof canWrite !== "boolean") { + // Fail closed: the endpoint sets all three only after authorizing. + ws.close(1008, "Missing flowId, userId or access decision"); return; } @@ -47,6 +50,7 @@ export function createYjsHandler() { close: () => ws.close(), }, userId, + canWrite, ); // Store cleanup function for later diff --git a/packages/collab/src/yjs-server.ts b/packages/collab/src/yjs-server.ts index 375b890c..d70e6bf3 100644 --- a/packages/collab/src/yjs-server.ts +++ b/packages/collab/src/yjs-server.ts @@ -31,6 +31,8 @@ export type Connection = { type ConnectionInfo = { awarenessClientIds: Set; // Track all awareness client IDs from this connection userId: string; + /** Viewers join read-only: their inbound doc writes are dropped. */ + canWrite: boolean; }; type Room = { @@ -58,16 +60,23 @@ export class YjsServer { // Connection Handling // -------------------------------------------------------------------------- + /** + * Attach a connection to a room. The caller is responsible for having + * authorized `userId` on `flowId` first — `canWrite` is required (not + * defaulted) so no caller can grant write access by omission. + */ async handleConnection( flowId: string, connection: Connection, userId: string, + canWrite: boolean, ): Promise<() => void> { const room = await this.getOrCreateRoom(flowId); room.connections.set(connection, { awarenessClientIds: new Set(), userId, + canWrite, }); console.log(`[YJS] Room ${flowId}: ${room.connections.size} connection(s)`); @@ -125,6 +134,16 @@ export class YjsServer { connection: Connection, decoder: decoding.Decoder, ): void { + // Read-only connections may only ask for our state (step 1). Step 2 and + // update messages both write into the doc, so drop them. An unknown + // connection has no ConnectionInfo and is treated as read-only. + if ( + !room.connections.get(connection)?.canWrite && + decoding.peekVarUint(decoder) !== syncProtocol.messageYjsSyncStep1 + ) { + return; + } + const encoder = encoding.createEncoder(); encoding.writeVarUint(encoder, MESSAGE_SYNC);