fix(collab): authorize the user for the flow on the Yjs websocket

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

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<typeof YjsServer>, 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<typeof YjsServer>;
+
+  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");
+  });
+});

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<WebSocketData>) => {
-      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

packages/collab/src/yjs-server.ts

@@ -31,6 +31,8 @@ export type Connection = {
 type ConnectionInfo = {
   awarenessClientIds: Set<number>; // 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);