fix(realtime): evict revoked collaborators from live workflow rooms via periodic read-access re-validation

Author: icecrasher321

apps/realtime/src/access-revalidation.test.ts

@@ -0,0 +1,181 @@
+/**
+ * @vitest-environment node
+ *
+ * Tests for the periodic read-access re-validation sweep. The security contract:
+ * a socket is evicted only when its role resolves to `null` (a confirmed
+ * revocation), and a transient failure never evicts a still-authorized socket.
+ */
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { mockResolveRole } = vi.hoisted(() => ({
+  mockResolveRole: vi.fn(),
+}))
+
+vi.mock('@/middleware/permissions', () => ({
+  resolveCurrentWorkflowRole: mockResolveRole,
+  ROLE_REVALIDATION_TTL_MS: 30_000,
+}))
+
+import { startAccessRevalidationSweep } from '@/access-revalidation'
+import type { IRoomManager, UserPresence } from '@/rooms'
+
+interface FakeSocket {
+  id: string
+  userId?: string
+  rooms: Set<string>
+  emit: ReturnType<typeof vi.fn>
+  leave: ReturnType<typeof vi.fn>
+}
+
+function makeSocket(id: string, userId: string | undefined, workflowId?: string): FakeSocket {
+  const rooms = new Set<string>([id])
+  if (workflowId) rooms.add(workflowId)
+  return { id, userId, rooms, emit: vi.fn(), leave: vi.fn() }
+}
+
+function makeManager(sockets: FakeSocket[], presence: Partial<UserPresence>[] = []) {
+  const socketMap = new Map(sockets.map((s) => [s.id, s]))
+  const manager = {
+    io: { sockets: { sockets: socketMap } },
+    isReady: () => true,
+    getWorkflowUsers: vi.fn().mockResolvedValue(presence),
+    removeUserFromRoom: vi.fn().mockResolvedValue(null),
+    broadcastPresenceUpdate: vi.fn().mockResolvedValue(undefined),
+  }
+  return manager as unknown as IRoomManager & {
+    getWorkflowUsers: ReturnType<typeof vi.fn>
+    removeUserFromRoom: ReturnType<typeof vi.fn>
+    broadcastPresenceUpdate: ReturnType<typeof vi.fn>
+  }
+}
+
+describe('access-revalidation sweep', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+  })
+
+  it('evicts a socket whose role has been revoked', async () => {
+    const socket = makeSocket('sock-1', 'user-1', 'wf-1')
+    const manager = makeManager([socket], [{ socketId: 'sock-1', role: 'read' }])
+    mockResolveRole.mockResolvedValue(null)
+
+    const sweep = startAccessRevalidationSweep(manager)
+    await sweep.runOnce()
+    sweep.stop()
+
+    expect(socket.emit).toHaveBeenCalledWith(
+      'access-revoked',
+      expect.objectContaining({ workflowId: 'wf-1' })
+    )
+    expect(socket.leave).toHaveBeenCalledWith('wf-1')
+    expect(manager.removeUserFromRoom).toHaveBeenCalledWith('sock-1', 'wf-1')
+    expect(manager.broadcastPresenceUpdate).toHaveBeenCalledWith('wf-1')
+  })
+
+  it('keeps a socket whose access is still valid', async () => {
+    const socket = makeSocket('sock-1', 'user-1', 'wf-1')
+    const manager = makeManager([socket], [{ socketId: 'sock-1', role: 'write' }])
+    mockResolveRole.mockResolvedValue('write')
+
+    const sweep = startAccessRevalidationSweep(manager)
+    await sweep.runOnce()
+    sweep.stop()
+
+    expect(socket.emit).not.toHaveBeenCalled()
+    expect(socket.leave).not.toHaveBeenCalled()
+    expect(manager.removeUserFromRoom).not.toHaveBeenCalled()
+  })
+
+  it('does not evict a downgraded-but-still-authorized socket', async () => {
+    const socket = makeSocket('sock-1', 'user-1', 'wf-1')
+    const manager = makeManager([socket], [{ socketId: 'sock-1', role: 'admin' }])
+    // Downgraded admin -> read still resolves to a non-null role: keep the reader.
+    mockResolveRole.mockResolvedValue('read')
+
+    const sweep = startAccessRevalidationSweep(manager)
+    await sweep.runOnce()
+    sweep.stop()
+
+    expect(socket.emit).not.toHaveBeenCalled()
+    expect(socket.leave).not.toHaveBeenCalled()
+  })
+
+  it('never evicts when re-validation throws (transient failure)', async () => {
+    const socket = makeSocket('sock-1', 'user-1', 'wf-1')
+    const manager = makeManager([socket], [{ socketId: 'sock-1', role: 'read' }])
+    mockResolveRole.mockRejectedValue(new Error('db unreachable'))
+
+    const sweep = startAccessRevalidationSweep(manager)
+    await sweep.runOnce()
+    sweep.stop()
+
+    expect(socket.emit).not.toHaveBeenCalled()
+    expect(socket.leave).not.toHaveBeenCalled()
+    expect(manager.removeUserFromRoom).not.toHaveBeenCalled()
+  })
+
+  it('passes the join-time presence role as the fallback', async () => {
+    const socket = makeSocket('sock-1', 'user-1', 'wf-1')
+    const manager = makeManager([socket], [{ socketId: 'sock-1', role: 'admin' }])
+    mockResolveRole.mockResolvedValue('admin')
+
+    const sweep = startAccessRevalidationSweep(manager)
+    await sweep.runOnce()
+    sweep.stop()
+
+    expect(mockResolveRole).toHaveBeenCalledWith('user-1', 'wf-1', 'admin')
+  })
+
+  it('evicts only the revoked socket, not co-members of the room', async () => {
+    const revoked = makeSocket('sock-1', 'user-1', 'wf-1')
+    const kept = makeSocket('sock-2', 'user-2', 'wf-1')
+    const manager = makeManager(
+      [revoked, kept],
+      [
+        { socketId: 'sock-1', role: 'read' },
+        { socketId: 'sock-2', role: 'write' },
+      ]
+    )
+    mockResolveRole.mockImplementation(async (userId: string) =>
+      userId === 'user-1' ? null : 'write'
+    )
+
+    const sweep = startAccessRevalidationSweep(manager)
+    await sweep.runOnce()
+    sweep.stop()
+
+    expect(revoked.leave).toHaveBeenCalledWith('wf-1')
+    expect(kept.leave).not.toHaveBeenCalled()
+    expect(kept.emit).not.toHaveBeenCalled()
+  })
+
+  it('skips unauthenticated sockets and sockets not in a workflow room', async () => {
+    const noUser = makeSocket('sock-1', undefined, 'wf-1')
+    const noRoom = makeSocket('sock-2', 'user-2')
+    const manager = makeManager([noUser, noRoom])
+
+    const sweep = startAccessRevalidationSweep(manager)
+    await sweep.runOnce()
+    sweep.stop()
+
+    expect(mockResolveRole).not.toHaveBeenCalled()
+    expect(noUser.leave).not.toHaveBeenCalled()
+    expect(noRoom.leave).not.toHaveBeenCalled()
+  })
+
+  it('still evaluates access when presence lookup fails (falls back safely)', async () => {
+    const socket = makeSocket('sock-1', 'user-1', 'wf-1')
+    const manager = makeManager([socket])
+    manager.getWorkflowUsers.mockRejectedValue(new Error('redis down'))
+    mockResolveRole.mockResolvedValue(null)
+
+    const sweep = startAccessRevalidationSweep(manager)
+    await sweep.runOnce()
+    sweep.stop()
+
+    // Presence unavailable → default fallback role, but the DB check still runs
+    // and a confirmed revocation still evicts.
+    expect(mockResolveRole).toHaveBeenCalledWith('user-1', 'wf-1', 'read')
+    expect(socket.leave).toHaveBeenCalledWith('wf-1')
+  })
+})

apps/realtime/src/access-revalidation.ts

@@ -0,0 +1,157 @@
+import { createLogger } from '@sim/logger'
+import type { AccessRevokedBroadcast } from '@sim/realtime-protocol/events'
+import type { AuthenticatedSocket } from '@/middleware/auth'
+import { ROLE_REVALIDATION_TTL_MS, resolveCurrentWorkflowRole } from '@/middleware/permissions'
+import type { IRoomManager } from '@/rooms'
+
+const logger = createLogger('AccessRevalidation')
+
+/**
+ * How often each pod re-validates live read access for its connected sockets.
+ *
+ * Coupled to {@link ROLE_REVALIDATION_TTL_MS} — the same per-pod role cache and
+ * TTL that already bound *write* revocation — so a collaborator whose workspace
+ * permission is removed loses live *reads* within a comparable window instead of
+ * retaining them until they disconnect. Detection latency is one cache TTL plus
+ * up to one sweep interval (the sweep keeps seeing the cached non-null role
+ * until it expires), so ~30s typical and bounded well under a minute worst case.
+ */
+export const ACCESS_REVALIDATION_SWEEP_INTERVAL_MS = ROLE_REVALIDATION_TTL_MS
+
+/**
+ * Non-null fallback used only when a socket has no presence entry to read its
+ * join-time role from. It is consumed by {@link resolveCurrentWorkflowRole} only
+ * on a transient DB failure with a cold cache, where returning a non-null role
+ * (never eviction) is the safe outcome during an outage.
+ */
+const FALLBACK_ROLE = 'read'
+
+export interface AccessRevalidationSweep {
+  /** Stop the periodic sweep (clears the interval). */
+  stop: () => void
+  /** Run a single sweep pass. Exposed for deterministic testing. */
+  runOnce: () => Promise<void>
+}
+
+/**
+ * Groups this pod's local sockets by the workflow room each has joined.
+ *
+ * The workflow room is derived from the socket's own `rooms` set (pod-local, no
+ * Redis round-trips): a socket joins exactly one workflow room, so its rooms are
+ * `{ ownSocketId, workflowId }`. Only local sockets are evaluated — sockets are
+ * sticky to a pod, so every socket is swept by exactly one pod using that pod's
+ * warm role cache (mirroring the per-pod reasoning of the write-path cache).
+ */
+function collectLocalMemberships(io: IRoomManager['io']): Map<string, AuthenticatedSocket[]> {
+  const byWorkflow = new Map<string, AuthenticatedSocket[]>()
+  for (const socket of io.sockets.sockets.values()) {
+    const authed = socket as AuthenticatedSocket
+    if (!authed.userId) continue
+    for (const room of socket.rooms) {
+      if (room === socket.id) continue
+      const existing = byWorkflow.get(room)
+      if (existing) existing.push(authed)
+      else byWorkflow.set(room, [authed])
+    }
+  }
+  return byWorkflow
+}
+
+/**
+ * Starts a per-pod loop that re-validates every connected socket's workspace
+ * role and evicts sockets whose access has been revoked, closing the read-side
+ * gap left by the join-only access check.
+ *
+ * Blip-safety: eviction fires *only* when {@link resolveCurrentWorkflowRole}
+ * returns `null`, which happens solely for a successful DB "no access" result or
+ * a previously-recorded revocation reused across a failure. A transient DB error
+ * against a still-authorized (or freshly-joined) user resolves to the last-known
+ * or fallback role, so a database blip never evicts anyone.
+ */
+export function startAccessRevalidationSweep(roomManager: IRoomManager): AccessRevalidationSweep {
+  const io = roomManager.io
+  let running = false
+
+  async function revokeSocket(socket: AuthenticatedSocket, workflowId: string): Promise<void> {
+    // Security-critical, pod-local, and synchronous: stop this socket receiving
+    // room broadcasts immediately, before any async bookkeeping that could fail.
+    const payload: AccessRevokedBroadcast = {
+      workflowId,
+      message: 'Your access to this workflow has been revoked',
+      timestamp: Date.now(),
+    }
+    socket.emit('access-revoked', payload)
+    socket.leave(workflowId)
+
+    logger.info(
+      `Revoked live access for user ${socket.userId} on workflow ${workflowId} (socket ${socket.id})`
+    )
+
+    // Best-effort room-state cleanup; failure here does not restore access.
+    await roomManager.removeUserFromRoom(socket.id, workflowId)
+    await roomManager.broadcastPresenceUpdate(workflowId)
+  }
+
+  async function runOnce(): Promise<void> {
+    const memberships = collectLocalMemberships(io)
+
+    for (const [workflowId, sockets] of memberships) {
+      // One presence read per active workflow supplies each socket's join-time
+      // role as the transient-failure fallback (a fallback never evicts).
+      const fallbackBySocket = new Map<string, string>()
+      try {
+        const presence = await roomManager.getWorkflowUsers(workflowId)
+        for (const entry of presence) {
+          fallbackBySocket.set(entry.socketId, entry.role)
+        }
+      } catch (error) {
+        logger.warn(`Failed to load presence for ${workflowId}; using default fallback role`, error)
+      }
+
+      for (const socket of sockets) {
+        const userId = socket.userId
+        if (!userId) continue
+
+        try {
+          const fallbackRole = fallbackBySocket.get(socket.id) ?? FALLBACK_ROLE
+          const role = await resolveCurrentWorkflowRole(userId, workflowId, fallbackRole)
+          if (role === null) {
+            await revokeSocket(socket, workflowId)
+          }
+        } catch (error) {
+          // Never evict on an unexpected error — only a definitive `null` role
+          // evicts, so a failure here leaves the socket's access intact.
+          logger.warn(
+            `Access re-validation failed for user ${userId} on workflow ${workflowId}; leaving membership intact`,
+            error
+          )
+        }
+      }
+    }
+  }
+
+  const timer = setInterval(() => {
+    if (running) {
+      logger.warn('Skipping access re-validation sweep; previous sweep still running')
+      return
+    }
+    running = true
+    runOnce()
+      .catch((error) => logger.error('Access re-validation sweep failed', error))
+      .finally(() => {
+        running = false
+      })
+  }, ACCESS_REVALIDATION_SWEEP_INTERVAL_MS)
+
+  // Do not keep the process alive solely for this timer.
+  timer.unref?.()
+
+  logger.info(
+    `Access re-validation sweep started (every ${ACCESS_REVALIDATION_SWEEP_INTERVAL_MS}ms)`
+  )
+
+  return {
+    stop: () => clearInterval(timer),
+    runOnce,
+  }
+}

apps/realtime/src/index.ts

@@ -1,6 +1,7 @@
 import { createServer } from 'http'
 import { createLogger } from '@sim/logger'
 import type { Server as SocketIOServer } from 'socket.io'
+import { startAccessRevalidationSweep } from '@/access-revalidation'
 import { createSocketIOServer, shutdownSocketIOAdapter } from '@/config/socket'
 import { assertSchemaCompatibility } from '@/database/preflight'
 import { env } from '@/env'
@@ -94,6 +95,10 @@ async function main() {
     setupAllHandlers(socket, roomManager)
   })
 
+  // Bound read-access staleness: periodically re-validate connected sockets and
+  // evict any whose workspace permission has been revoked, matching the write path.
+  const accessRevalidation = startAccessRevalidationSweep(roomManager)
+
   await assertSchemaCompatibility()
 
   httpServer.listen(PORT, '0.0.0.0', () => {
@@ -104,6 +109,8 @@ async function main() {
   const shutdown = async () => {
     logger.info('Shutting down Socket.IO server...')
 
+    accessRevalidation.stop()
+
     try {
       await roomManager.shutdown()
       logger.info('RoomManager shutdown complete')

apps/realtime/src/middleware/permissions.ts

@@ -85,11 +85,13 @@ export function checkRolePermission(
 }
 
 /**
- * TTL for the per-pod role cache backing live re-validation of mutating operations.
- * Bounds how long a revoked or downgraded collaborator can retain write access on an
- * already-connected socket.
+ * TTL for the per-pod role cache backing live re-validation. It gates both the
+ * mutating-operation checks ({@link checkWorkflowOperationPermission}) and the
+ * periodic read-access sweep (`access-revalidation.ts`), so a revoked or
+ * downgraded collaborator loses write access — and live reads — on an
+ * already-connected socket within a bounded window rather than until disconnect.
  */
-const ROLE_REVALIDATION_TTL_MS = 30_000
+export const ROLE_REVALIDATION_TTL_MS = 30_000
 
 /** Soft cap on cached entries before an opportunistic purge of expired ones runs. */
 const MAX_ROLE_CACHE_ENTRIES = 5_000