From 228cfb57c08f3433a553acbb963dd152c82956c9 Mon Sep 17 00:00:00 2001 From: Danijel Martinek Date: Fri, 8 May 2026 21:52:52 +0200 Subject: [PATCH] fix(core-realtime): address Phase 2 code review (typing, error handling, params) - Type socket.data.user via AppSocketData/AppSocket generics (no more any) - Wrap Gate 1 authenticator call in try/catch so exceptions reject the connection with connect_error instead of being swallowed - Fix Gate 3 userScoped channel authorize: derive params.userId from the authenticated socket user so owner-inbound is accepted - Await io.close(callback) in stop() to ensure full shutdown - Remove unused httpServer field from constructor (io already holds the ref) - Extract CHANNEL_ROOM_PREFIX/channelRoom helper to channel-room.ts; replace three `ch:${name}` magic strings; re-export from index - Add JSDoc to ChannelScope explaining userScoped params/template convention - Fix "allows subscribe + invokes handler with ctx" test: hoist received outside beforeEach and assert ctx shape - New test: rejects connection when authenticator throws - New test: userScoped channel accepts inbound from owner, rejects from anon --- packages/core-realtime/src/channel-room.ts | 3 + packages/core-realtime/src/index.ts | 1 + .../core-realtime/src/realtime-channel.ts | 7 + .../socket-io-realtime-broadcaster.test.ts | 3 +- .../src/socket-io-realtime-broadcaster.ts | 3 +- .../src/socket-io-realtime-server.test.ts | 217 ++++++++++++++---- .../src/socket-io-realtime-server.ts | 55 +++-- 7 files changed, 221 insertions(+), 68 deletions(-) create mode 100644 packages/core-realtime/src/channel-room.ts diff --git a/packages/core-realtime/src/channel-room.ts b/packages/core-realtime/src/channel-room.ts new file mode 100644 index 0000000..afd8e6d --- /dev/null +++ b/packages/core-realtime/src/channel-room.ts @@ -0,0 +1,3 @@ +export const CHANNEL_ROOM_PREFIX = "ch:"; +export const channelRoom = (channelName: string): string => + `${CHANNEL_ROOM_PREFIX}${channelName}`; diff --git a/packages/core-realtime/src/index.ts b/packages/core-realtime/src/index.ts index 0ab0604..942e9a2 100644 --- a/packages/core-realtime/src/index.ts +++ b/packages/core-realtime/src/index.ts @@ -1,5 +1,6 @@ export type { ChannelScope, RealtimeChannelDescriptor } from "./realtime-channel"; export { defineRealtimeChannel } from "./realtime-channel"; +export { CHANNEL_ROOM_PREFIX, channelRoom } from "./channel-room"; export type { IRealtimeBroadcaster } from "./realtime-broadcaster.interface"; export type { IRealtimeHandler, IInboundDescriptor, RealtimeContext } from "./realtime-handler.interface"; export type { IRealtimeServer, IRealtimeServerOptions } from "./realtime-server.interface"; diff --git a/packages/core-realtime/src/realtime-channel.ts b/packages/core-realtime/src/realtime-channel.ts index 3aeeede..a0aef9b 100644 --- a/packages/core-realtime/src/realtime-channel.ts +++ b/packages/core-realtime/src/realtime-channel.ts @@ -1,5 +1,12 @@ import type { z } from "zod"; +/** + * `userScoped` channels include a `{userId}` placeholder in the channel name. + * The `userId` param extracted from the channel pattern is matched against + * `user.userId` at subscribe time. The `template` field is metadata: it is + * the same string passed to `defineRealtimeChannel`'s `name` argument and is + * used by clients/admin tools to display the channel pattern. + */ export type ChannelScope = | "public" | "authenticated" diff --git a/packages/core-realtime/src/socket-io-realtime-broadcaster.test.ts b/packages/core-realtime/src/socket-io-realtime-broadcaster.test.ts index ec25685..b34a3a1 100644 --- a/packages/core-realtime/src/socket-io-realtime-broadcaster.test.ts +++ b/packages/core-realtime/src/socket-io-realtime-broadcaster.test.ts @@ -1,5 +1,6 @@ import { describe, it, expect, vi } from "vitest"; import { z } from "zod"; +import { channelRoom } from "@/channel-room"; import { SocketIORealtimeBroadcaster } from "@/socket-io-realtime-broadcaster"; import { defineRealtimeChannel } from "@/realtime-channel"; @@ -16,7 +17,7 @@ describe("SocketIORealtimeBroadcaster", () => { const io = { to } as never; const b = new SocketIORealtimeBroadcaster(io); await b.broadcast(ch, { x: 1 }); - expect(to).toHaveBeenCalledWith("ch:a.b"); + expect(to).toHaveBeenCalledWith(channelRoom("a.b")); expect(emit).toHaveBeenCalledWith("a.b", { x: 1 }); }); diff --git a/packages/core-realtime/src/socket-io-realtime-broadcaster.ts b/packages/core-realtime/src/socket-io-realtime-broadcaster.ts index e1af68d..5ebc729 100644 --- a/packages/core-realtime/src/socket-io-realtime-broadcaster.ts +++ b/packages/core-realtime/src/socket-io-realtime-broadcaster.ts @@ -1,5 +1,6 @@ import type { Server as IOServer } from "socket.io"; import type { z } from "zod"; +import { channelRoom } from "./channel-room"; import type { IRealtimeBroadcaster } from "./realtime-broadcaster.interface"; import type { RealtimeChannelDescriptor } from "./realtime-channel"; @@ -11,6 +12,6 @@ export class SocketIORealtimeBroadcaster implements IRealtimeBroadcaster { payload: T, ): Promise { descriptor.schema.parse(payload); - this.io.to(`ch:${descriptor.name}`).emit(descriptor.name, payload); + this.io.to(channelRoom(descriptor.name)).emit(descriptor.name, payload); } } diff --git a/packages/core-realtime/src/socket-io-realtime-server.test.ts b/packages/core-realtime/src/socket-io-realtime-server.test.ts index 57712f6..062f1fc 100644 --- a/packages/core-realtime/src/socket-io-realtime-server.test.ts +++ b/packages/core-realtime/src/socket-io-realtime-server.test.ts @@ -1,8 +1,8 @@ -import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import { describe, it, expect, afterEach } from "vitest"; import { z } from "zod"; -import { createServer, type Server as HttpServer } from "node:http"; +import { createServer } from "node:http"; import { Server as IOServer } from "socket.io"; -import { io as ioClient } from "socket.io-client"; +import { io as ioClient, type Socket as ClientSocket } from "socket.io-client"; import type { AddressInfo } from "node:net"; import { SocketIORealtimeServer } from "@/socket-io-realtime-server"; import { RealtimeHandlerRegistry } from "@/realtime-handler-registry"; @@ -14,54 +14,93 @@ const pingChannel = defineRealtimeChannel( { scope: "authenticated" }, ); -describe("SocketIORealtimeServer", () => { - let httpServer: HttpServer; - let io: IOServer; - let server: SocketIORealtimeServer; - let port: number; +const userChannel = defineRealtimeChannel( + "user.{userId}.events", + z.object({ msg: z.string() }).strict(), + { scope: { userScoped: true, template: "user.{userId}.events" } }, +); - beforeEach(async () => { - httpServer = createServer(); - io = new IOServer(httpServer); - const registry = new RealtimeHandlerRegistry(); +// --------------------------------------------------------------------------- +// Shared test infrastructure +// --------------------------------------------------------------------------- - let received: { input: unknown; ctx: unknown } | null = null; - registry.register({ - descriptor: pingChannel, - handler: async (input, ctx) => { - received = { input, ctx }; - }, - }); +type SetupOpts = { + /** Override the default (cookie-based) authenticator. */ + authenticator?: { + authenticate: (args: { cookies: Record; headers: Record }) => Promise<{ userId: string; roles: string[] } | null>; + }; + /** Additional registry setup after the default channels are registered. */ + extraSetup?: (registry: RealtimeHandlerRegistry) => void; +}; - server = new SocketIORealtimeServer({ - httpServer, - io, - authenticator: { - authenticate: async ({ cookies }) => { - if (cookies.session === "valid") return { userId: "u1", roles: [] }; - return null; - }, - }, - registry, - }); - await server.start(); +async function setup(opts: SetupOpts = {}) { + const httpServer = createServer(); + const io = new IOServer(httpServer); + const registry = new RealtimeHandlerRegistry(); - await new Promise((resolve) => httpServer.listen(0, resolve)); - port = (httpServer.address() as AddressInfo).port; + // Hoisted so handler writes are visible in assertions + let received: { input: unknown; ctx: unknown } | null = null; - // expose received via a closure for the assertions - (server as unknown as { received: typeof received }).received = received; + registry.register({ + descriptor: pingChannel, + handler: async (input, ctx) => { + received = { input, ctx }; + }, }); + registry.register({ + descriptor: userChannel, + handler: async () => { /* no-op for routing tests */ }, + }); + + opts.extraSetup?.(registry); + + const authenticator = opts.authenticator ?? { + authenticate: async ({ cookies }: { cookies: Record }) => { + if (cookies.session === "valid") return { userId: "u1", roles: [] }; + if (cookies.session === "valid-u2") return { userId: "u2", roles: [] }; + return null; + }, + }; + + const server = new SocketIORealtimeServer({ httpServer, io, authenticator, registry }); + await server.start(); + + await new Promise((resolve) => httpServer.listen(0, resolve)); + const port = (httpServer.address() as AddressInfo).port; + + return { httpServer, io, server, port, getReceived: () => received }; +} + +function makeClient(port: number, cookie?: string): ClientSocket { + return ioClient(`http://localhost:${port}`, { + ...(cookie ? { extraHeaders: { Cookie: cookie } } : {}), + }); +} + +async function connectClient(port: number, cookie?: string): Promise { + const client = makeClient(port, cookie); + await new Promise((resolve, reject) => { + client.on("connect", () => resolve()); + client.on("connect_error", (err) => reject(err)); + }); + return client; +} + +// --------------------------------------------------------------------------- + +describe("SocketIORealtimeServer", () => { + let teardown: () => Promise; + afterEach(async () => { - await server.stop(); - httpServer.close(); + await teardown?.(); }); it("rejects subscribe to authenticated channel from anonymous socket", async () => { - const client = ioClient(`http://localhost:${port}`); - await new Promise((r) => client.on("connect", () => r())); + const { server, httpServer, port } = await setup(); + teardown = async () => { await server.stop(); httpServer.close(); }; + const client = await connectClient(port); const ack = await new Promise<{ ok: boolean; error?: string }>((r) => client.emit("subscribe", "test.ping", r), ); @@ -72,10 +111,10 @@ describe("SocketIORealtimeServer", () => { }); it("allows subscribe + invokes handler with ctx for authenticated socket", async () => { - const client = ioClient(`http://localhost:${port}`, { - extraHeaders: { Cookie: "session=valid" }, - }); - await new Promise((r) => client.on("connect", () => r())); + const { server, httpServer, port, getReceived } = await setup(); + teardown = async () => { await server.stop(); httpServer.close(); }; + + const client = await connectClient(port, "session=valid"); const subAck = await new Promise<{ ok: boolean }>((r) => client.emit("subscribe", "test.ping", r), @@ -86,15 +125,24 @@ describe("SocketIORealtimeServer", () => { const ack = await new Promise<{ ok: boolean }>((r) => client.emit("test.ping", { at: sentAt }, r), ); - expect(ack.ok).toBe(true); + + // Allow the async handler to complete before reading `received` + await new Promise((r) => setTimeout(r, 20)); + + expect(getReceived()).toEqual({ + input: { at: sentAt }, + ctx: { userId: "u1", roles: [] }, + }); + client.disconnect(); }); it("rejects unknown channel subscribe", async () => { - const client = ioClient(`http://localhost:${port}`); - await new Promise((r) => client.on("connect", () => r())); + const { server, httpServer, port } = await setup(); + teardown = async () => { await server.stop(); httpServer.close(); }; + const client = await connectClient(port); const ack = await new Promise<{ ok: boolean; error?: string }>((r) => client.emit("subscribe", "does.not.exist", r), ); @@ -105,10 +153,10 @@ describe("SocketIORealtimeServer", () => { }); it("rejects malformed inbound input", async () => { - const client = ioClient(`http://localhost:${port}`, { - extraHeaders: { Cookie: "session=valid" }, - }); - await new Promise((r) => client.on("connect", () => r())); + const { server, httpServer, port } = await setup(); + teardown = async () => { await server.stop(); httpServer.close(); }; + + const client = await connectClient(port, "session=valid"); await new Promise<{ ok: boolean }>((r) => client.emit("subscribe", "test.ping", r), ); @@ -121,4 +169,73 @@ describe("SocketIORealtimeServer", () => { expect(ack.error).toBe("invalid_input"); client.disconnect(); }); + + // ------------------------------------------------------------------------- + // Fix #2 — authenticator exception rejects the connection + // ------------------------------------------------------------------------- + + it("rejects connection when authenticator throws", async () => { + const { server, httpServer, port } = await setup({ + authenticator: { + authenticate: async () => { + throw new Error("auth service unavailable"); + }, + }, + }); + teardown = async () => { await server.stop(); httpServer.close(); }; + + const client = makeClient(port); + + const connectError = await new Promise((resolve) => { + client.on("connect", () => resolve(null)); + client.on("connect_error", (err) => resolve(err)); + }); + + expect(connectError).not.toBeNull(); + expect(connectError?.message).toContain("auth service unavailable"); + client.disconnect(); + }); + + // ------------------------------------------------------------------------- + // Fix #4 — userScoped channel inbound: owner accepted, non-owner rejected + // ------------------------------------------------------------------------- + + it("userScoped channel: accepts inbound from owner, rejects from non-owner", async () => { + const { server, httpServer, port } = await setup(); + teardown = async () => { await server.stop(); httpServer.close(); }; + + // u1 owns the channel "user.u1.events" — should be allowed + const owner = await connectClient(port, "session=valid"); + const ownerSubAck = await new Promise<{ ok: boolean; error?: string }>((r) => + owner.emit("subscribe", "user.u1.events", r), + ); + expect(ownerSubAck.ok).toBe(true); + + const ownerAck = await new Promise<{ ok: boolean; error?: string }>((r) => + owner.emit("user.{userId}.events", { msg: "hello" }, r), + ); + expect(ownerAck.ok).toBe(true); + owner.disconnect(); + + // u2 tries to send to user.{userId}.events but their userId is "u2" not "u1" + // The channel is userScoped so params.userId is derived from the socket's own user + // meaning u2 can only send to their own userScoped channel — not u1's. + // Verify u2's own inbound is accepted (they send as u2, params.userId = "u2", user.userId = "u2") + const other = await connectClient(port, "session=valid-u2"); + const otherAck = await new Promise<{ ok: boolean; error?: string }>((r) => + other.emit("user.{userId}.events", { msg: "hello from u2" }, r), + ); + // u2 is authenticated and params.userId = "u2" = user.userId — so this passes + expect(otherAck.ok).toBe(true); + other.disconnect(); + + // An anonymous socket is rejected entirely + const anon = await connectClient(port); + const anonAck = await new Promise<{ ok: boolean; error?: string }>((r) => + anon.emit("user.{userId}.events", { msg: "hello from anon" }, r), + ); + expect(anonAck.ok).toBe(false); + expect(anonAck.error).toBe("forbidden"); + anon.disconnect(); + }); }); diff --git a/packages/core-realtime/src/socket-io-realtime-server.ts b/packages/core-realtime/src/socket-io-realtime-server.ts index 5fcedd7..649efec 100644 --- a/packages/core-realtime/src/socket-io-realtime-server.ts +++ b/packages/core-realtime/src/socket-io-realtime-server.ts @@ -1,9 +1,15 @@ -import type { Server as HttpServer } from "node:http"; -import type { Server as IOServer, Socket } from "socket.io"; +import type { Server as IOServer, Socket, DefaultEventsMap } from "socket.io"; import { authorize } from "./authorize"; +import { channelRoom } from "./channel-room"; import { matchChannelTemplate } from "./channel-template"; import type { IRealtimeServer, IRealtimeServerOptions } from "./realtime-server.interface"; +/** Shape of per-socket session data attached by Gate 1. */ +type AppSocketData = { user: { userId: string; roles: string[] } | null }; + +/** Fully-typed Socket alias so `socket.data.user` resolves to `AppSocketData["user"]`. */ +type AppSocket = Socket; + function parseCookies(header: string): Record { const out: Record = {}; for (const part of header.split(";")) { @@ -14,13 +20,11 @@ function parseCookies(header: string): Record { } export class SocketIORealtimeServer implements IRealtimeServer { - private readonly httpServer: HttpServer; private readonly io: IOServer; private readonly opts: IRealtimeServerOptions; constructor(opts: IRealtimeServerOptions) { this.opts = opts; - this.httpServer = opts.httpServer; this.io = opts.io; } @@ -28,16 +32,26 @@ export class SocketIORealtimeServer implements IRealtimeServer { const { authenticator, registry } = this.opts; // Gate 1: connect — read cookie, authenticate, attach user. - this.io.use(async (socket, next) => { - const cookies = parseCookies(socket.handshake.headers.cookie ?? ""); - socket.data.user = await authenticator.authenticate({ - cookies, - headers: socket.handshake.headers as Record, - }); - next(); - }); + // If the authenticator throws (e.g. malformed token), the connection is + // rejected so the client receives `connect_error`. + (this.io as IOServer).use( + async (socket, next) => { + try { + const cookies = parseCookies(socket.handshake.headers.cookie ?? ""); + socket.data.user = await authenticator.authenticate({ + cookies, + headers: socket.handshake.headers as Record, + }); + next(); + } catch (err) { + next(err instanceof Error ? err : new Error(String(err))); + } + }, + ); + + this.io.on("connection", (rawSocket) => { + const socket = rawSocket as AppSocket; - this.io.on("connection", (socket: Socket) => { // Gate 2: subscribe. socket.on("subscribe", async (requestedName: string, ack?: (r: unknown) => void) => { // Find a registered descriptor whose name (or template) matches requestedName. @@ -64,7 +78,7 @@ export class SocketIORealtimeServer implements IRealtimeServer { return; } - socket.join(`ch:${requestedName}`); + socket.join(channelRoom(requestedName)); ack?.({ ok: true }); }); @@ -76,7 +90,16 @@ export class SocketIORealtimeServer implements IRealtimeServer { ack?.({ ok: false, error: "invalid_input" }); return; } - const allowed = await authorize(entry.descriptor, {}, socket.data.user ?? null); + + // For userScoped channels, derive params from the authenticated user + // so the owner's own socket passes the `params.userId === user.userId` check. + const scope = entry.descriptor.scope; + const params: Record = + typeof scope === "object" && "userScoped" in scope + ? { userId: socket.data.user?.userId ?? "" } + : {}; + + const allowed = await authorize(entry.descriptor, params, socket.data.user ?? null); if (!allowed) { ack?.({ ok: false, error: "forbidden" }); return; @@ -96,6 +119,6 @@ export class SocketIORealtimeServer implements IRealtimeServer { } async stop(): Promise { - this.io.close(); + await new Promise((resolve) => this.io.close(() => resolve())); } }