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);