release: WRNexusJS 0.7.0

packages/core/src/realtime.ts

@@ -100,7 +100,28 @@ export interface RoomAuthInfo {
   headers: Headers;
 }
 
+export interface RealtimeSecurityOptions {
+  /** Maximum inbound or outbound serialized message size. Defaults to 64 KiB. */
+  maxMessageBytes?: number;
+  /** Maximum messages accepted per connection per rolling second. Defaults to 30. */
+  maxMessagesPerSecond?: number;
+  /** Maximum live connections in one room. Defaults to 1,000. */
+  maxConnectionsPerRoom?: number;
+  /** Maximum connections for one authenticated user in a room. Defaults to 10. */
+  maxConnectionsPerUser?: number;
+  /** Reject anonymous connections before onConnect. */
+  requireUser?: boolean;
+  /** Maximum nested JSON depth. Defaults to 32. */
+  maxJsonDepth?: number;
+  /** Optional message schema/authorization predicate. */
+  validateMessage?(message: unknown, client: RoomClient): boolean | Promise<boolean>;
+  /** Called when a connection is rejected or closed for a policy violation. */
+  onViolation?(reason: string, client?: RoomClient): void;
+}
+
 export interface RoomHandlers<TData = Record<string, unknown>> {
+  /** Per-room abuse and payload controls. */
+  security?: RealtimeSecurityOptions;
   /**
    * Gate the connection BEFORE it is accepted. Return false to reject the
    * upgrade with 403 (e.g. `authorize: (info) => !!info.user` to require auth).
@@ -138,6 +159,8 @@ export function isRoomDefinition(value: unknown): value is RoomDefinition {
 
 interface Conn {
   id: string;
+  messageWindowStartedAt: number;
+  messageCount: number;
   user?: string;
   data: Record<string, unknown>;
   query: Record<string, string>;
@@ -179,6 +202,10 @@ export interface RealtimeBridge {
   publish(envelope: RealtimeEnvelope): void;
 }
 
+export interface RealtimeRegistryOptions extends RealtimeSecurityOptions {
+  now?: () => number;
+}
+
 export interface RealtimeRegistry {
   open(socket: RawSocket, meta: RealtimeConnectMeta): void | Promise<void>;
   message(socket: RawSocket, raw: string | Uint8Array): void | Promise<void>;
@@ -191,16 +218,59 @@ export interface RealtimeRegistry {
   size(): number;
 }
 
+const DANGEROUS_REALTIME_KEYS = new Set(["__proto__", "prototype", "constructor"]);
+
+function assertRealtimePayload(value: unknown, maxDepth: number, depth = 0): void {
+  if (depth > maxDepth) throw new Error("Realtime payload nesting limit exceeded.");
+  if (!value || typeof value !== "object") return;
+  for (const [key, child] of Object.entries(value as Record<string, unknown>)) {
+    if (DANGEROUS_REALTIME_KEYS.has(key)) throw new Error(`Dangerous realtime key '${key}'.`);
+    assertRealtimePayload(child, maxDepth, depth + 1);
+  }
+}
+
+function byteLength(value: string | Uint8Array): number {
+  return typeof value === "string" ? new TextEncoder().encode(value).byteLength : value.byteLength;
+}
+
 function serialize(message: unknown): string {
   return typeof message === "string" ? message : JSON.stringify(message);
 }
 
+function mergedRealtimeSecurity(
+  globalOptions: RealtimeRegistryOptions,
+  room: RoomDefinition,
+): Required<
+  Pick<
+    RealtimeSecurityOptions,
+    | "maxMessageBytes"
+    | "maxMessagesPerSecond"
+    | "maxConnectionsPerRoom"
+    | "maxConnectionsPerUser"
+    | "requireUser"
+    | "maxJsonDepth"
+  >
+> &
+  RealtimeSecurityOptions {
+  return {
+    maxMessageBytes: 64 * 1024,
+    maxMessagesPerSecond: 30,
+    maxConnectionsPerRoom: 1_000,
+    maxConnectionsPerUser: 10,
+    requireUser: false,
+    maxJsonDepth: 32,
+    ...globalOptions,
+    ...room.handlers.security,
+  };
+}
+
 /** Create the registry that maps sockets ↔ rooms and drives room handlers. */
-export function createRealtimeRegistry(): RealtimeRegistry {
+export function createRealtimeRegistry(options: RealtimeRegistryOptions = {}): RealtimeRegistry {
   const rooms = new Map<string, RoomImpl>();
   const bySocket = new Map<RawSocket, Conn>();
   let bridge: RealtimeBridge | null = null;
   let applyingRemote = false; // true while delivering a peer envelope (no re-publish)
+  const now = options.now ?? Date.now;
 
   const publish = (envelope: RealtimeEnvelope): void => {
     if (bridge && !applyingRemote) bridge.publish(envelope);
@@ -208,6 +278,17 @@ export function createRealtimeRegistry(): RealtimeRegistry {
 
   const send = (conn: Conn | undefined, payload: string): void => {
     if (!conn) return;
+    const room = rooms.get(conn.roomName);
+    const policy = room
+      ? mergedRealtimeSecurity(options, room.def)
+      : ({ maxMessageBytes: options.maxMessageBytes ?? 64 * 1024 } as ReturnType<
+          typeof mergedRealtimeSecurity
+        >);
+    if (byteLength(payload) > policy.maxMessageBytes) {
+      policy.onViolation?.("outbound-message-too-large", conn.client);
+      conn.socket.close(1009, "Message too large");
+      return;
+    }
     try {
       conn.socket.send(payload);
     } catch {
@@ -295,13 +376,31 @@ export function createRealtimeRegistry(): RealtimeRegistry {
 
   return {
     async open(socket, meta) {
+      const policy = mergedRealtimeSecurity(options, meta.def);
+      if (policy.requireUser && !meta.user) {
+        policy.onViolation?.("authentication-required");
+        socket.close(1008, "Authentication required");
+        return;
+      }
       let room = rooms.get(meta.room);
       if (!room) {
         room = { name: meta.room, state: {}, def: meta.def, conns: new Map(), users: new Map() };
         rooms.set(meta.room, room);
       }
+      if (room.conns.size >= policy.maxConnectionsPerRoom) {
+        policy.onViolation?.("room-connection-limit");
+        socket.close(1013, "Room is at capacity");
+        return;
+      }
+      if (meta.user && (room.users.get(meta.user)?.size ?? 0) >= policy.maxConnectionsPerUser) {
+        policy.onViolation?.("user-connection-limit");
+        socket.close(1008, "Too many connections");
+        return;
+      }
       const conn: Conn = {
         id: randomId(),
+        messageWindowStartedAt: now(),
+        messageCount: 0,
         data: {},
         query: meta.query ?? {},
         socket,
@@ -320,6 +419,23 @@ export function createRealtimeRegistry(): RealtimeRegistry {
       if (!conn) return;
       const room = rooms.get(conn.roomName);
       if (!room) return;
+      const policy = mergedRealtimeSecurity(options, room.def);
+      if (byteLength(raw) > policy.maxMessageBytes) {
+        policy.onViolation?.("inbound-message-too-large", conn.client);
+        socket.close(1009, "Message too large");
+        return;
+      }
+      const timestamp = now();
+      if (timestamp - conn.messageWindowStartedAt >= 1_000) {
+        conn.messageWindowStartedAt = timestamp;
+        conn.messageCount = 0;
+      }
+      conn.messageCount += 1;
+      if (conn.messageCount > policy.maxMessagesPerSecond) {
+        policy.onViolation?.("message-rate-limit", conn.client);
+        socket.close(1008, "Message rate exceeded");
+        return;
+      }
       const text = typeof raw === "string" ? raw : new TextDecoder().decode(raw);
       let message: unknown;
       try {
@@ -327,6 +443,18 @@ export function createRealtimeRegistry(): RealtimeRegistry {
       } catch {
         message = text;
       }
+      try {
+        assertRealtimePayload(message, policy.maxJsonDepth);
+      } catch {
+        policy.onViolation?.("invalid-message-shape", conn.client);
+        socket.close(1008, "Invalid message");
+        return;
+      }
+      if (policy.validateMessage && !(await policy.validateMessage(message, conn.client))) {
+        policy.onViolation?.("message-validation-failed", conn.client);
+        socket.close(1008, "Message rejected");
+        return;
+      }
       await room.def.handlers.onMessage?.(conn.client, message);
     },