fix(express,node): prevent process crashes from ws frames and lock heartbeat

packages/express/ws-server.test.ts

@@ -2,7 +2,7 @@ import { describe, it } from 'node:test'
 import { strict as assert } from 'assert'
 import { IncomingMessage } from 'node:http'
 import { Socket } from 'node:net'
-import { subscribeDecision, subscribeForRequest } from './ws-server.js'
+import { subscribeDecision, subscribeForRequest, parseFrame } from './ws-server.js'
 import { session } from './session.js'
 import type { SessionState } from '@data-fair/lib-common-types/session/index.js'
 
@@ -126,3 +126,41 @@ describe('ws-server subscribeForRequest', () => {
     })
   })
 })
+
+describe('ws-server parseFrame', () => {
+  // regression: a bare "null" frame passed the typeof guard and then crashed the message handler
+  it('a null frame is rejected with a 400, not a throw', () => {
+    const r = parseFrame('null') as any
+    assert.equal(r.error.status, 400)
+    assert.equal(r.error.data, '"channel" is required')
+  })
+
+  it('other non-object frames are rejected without throwing', () => {
+    for (const frame of ['123', '"str"', 'true', '[]']) {
+      const r = parseFrame(frame) as any
+      assert.equal(r.error.status, 400, `frame ${frame}`)
+    }
+  })
+
+  it('invalid JSON is rejected with a 400', () => {
+    const r = parseFrame('{not json') as any
+    assert.equal(r.error.status, 400)
+  })
+
+  it('a frame without channel is rejected', () => {
+    const r = parseFrame('{"type":"subscribe"}') as any
+    assert.equal(r.error.data, '"channel" is required')
+  })
+
+  it('a frame with an invalid type is rejected but keeps the channel', () => {
+    const r = parseFrame('{"channel":"ch","type":"nope"}') as any
+    assert.equal(r.error.status, 400)
+    assert.equal(r.error.channel, 'ch')
+  })
+
+  it('a valid subscribe frame is accepted', () => {
+    const r = parseFrame('{"channel":"ch","type":"subscribe"}') as any
+    assert.equal(r.message.channel, 'ch')
+    assert.equal(r.message.type, 'subscribe')
+  })
+})

packages/express/ws-server.ts

@@ -82,6 +82,24 @@ export const subscribeForRequest = async (
 }
 
 let stopped = false
+// parse and validate a raw client frame. Must never throw on arbitrary input:
+// a bare "null" used to pass the typeof guard and then crash the handler.
+export const parseFrame = (str: string): { error: { type: 'error', status: number, data: any, channel?: string } } | { message: { channel: string, type: string, [k: string]: any } } => {
+  let message
+  try {
+    message = JSON.parse(str)
+  } catch (err: any) {
+    return { error: { type: 'error', status: 400, data: err.message ?? err } }
+  }
+  if (!message || typeof message !== 'object' || !message.channel) {
+    return { error: { type: 'error', status: 400, data: '"channel" is required' } }
+  }
+  if (!message.type || ['subscribe', 'unsubscribe'].indexOf(message.type) === -1) {
+    return { error: { type: 'error', channel: message.channel, status: 400, data: 'type should be "subscribe" or "unsubscribe"' } }
+  }
+  return { message }
+}
+
 export const start = async (
   server: Server,
   db: Db,
@@ -99,20 +117,10 @@ export const start = async (
     ws.on('message', async str => {
       if (stopped) return
 
-      let message
+      const parsed = parseFrame(str.toString())
+      if ('error' in parsed) return ws.send(JSON.stringify(parsed.error))
+      const message = parsed.message
       try {
-        message = JSON.parse(str.toString())
-      } catch (err: any) {
-        const errorMessage = { type: 'error', status: 400, data: err.message ?? err }
-        return ws.send(JSON.stringify(errorMessage))
-      }
-      try {
-        if (typeof message !== 'object' || !message.channel) {
-          return ws.send(JSON.stringify({ type: 'error', status: 400, data: '"channel" is required' }))
-        }
-        if (!message.type || ['subscribe', 'unsubscribe'].indexOf(message.type) === -1) {
-          return ws.send(JSON.stringify({ type: 'error', channel: message.channel, status: 400, data: 'type should be "subscribe" or "unsubscribe"' }))
-        }
         if (message.type === 'subscribe') {
           const decision = await subscribeForRequest(req, message.channel, message, canSubscribe, options.channelScopes)
           if (!decision.ok) {
@@ -129,7 +137,7 @@ export const start = async (
         }
       } catch (err: any) {
         const status = err.status ?? err.statusCode ?? 500
-        const errorMessage: Message = { type: 'error', status, data: err.message ?? err, channel: message.channel }
+        const errorMessage: Message = { type: 'error', status, data: err.message ?? err, channel: message?.channel }
         if (status >= 500) internalError('ws-error', err)
         return ws.send(JSON.stringify(errorMessage))
       }

packages/node/locks.ts

@@ -2,6 +2,7 @@ import type { Db } from 'mongodb'
 import { hostname } from 'node:os'
 import { randomBytes } from 'node:crypto'
 import Debug from 'debug'
+import { internalError } from './observer.js'
 
 const debug = Debug('locks')
 
@@ -35,8 +36,10 @@ export class Locks {
     }
 
     // prolongate lock acquired by this process while it is still active
+    // a mongo error here must not become an unhandled rejection (it would crash the process on any failover)
     this.interval = setInterval(() => {
       this.collection.updateMany({ pid: this.pid }, { $currentDate: { updatedAt: true } })
+        .catch(err => internalError('locks-heartbeat', err))
     }, (ttl / 2) * 1000)
   }