From 14ba9a90c6c83747ecd4bfeb76c4badeed86625d Mon Sep 17 00:00:00 2001 From: BatLeDev Date: Mon, 19 Oct 2026 18:00:44 +0200 Subject: [PATCH] fix(express,node): prevent process crashes from ws frames and lock heartbeat A bare "null" ws frame passed the typeof guard and then threw while reading message.channel, both in the guard and again in the catch, so the rejection escaped the async message handler and crashed the process. Any unauthenticated client could send one frame to take the API down, on every restart. Extract frame parsing/validation into a pure parseFrame() that never throws on arbitrary input, with regression tests. Also guard the locks heartbeat setInterval with a .catch: an un-awaited updateMany rejection (e.g. on a mongo failover) was an unhandled rejection that crashed every pod. --- packages/express/ws-server.test.ts | 40 +++++++++++++++++++++++++++++- packages/express/ws-server.ts | 36 ++++++++++++++++----------- packages/node/locks.ts | 3 +++ 3 files changed, 64 insertions(+), 15 deletions(-) diff --git a/packages/express/ws-server.test.ts b/packages/express/ws-server.test.ts index 891d7af..dde2650 100644 --- a/packages/express/ws-server.test.ts +++ b/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') + }) +}) diff --git a/packages/express/ws-server.ts b/packages/express/ws-server.ts index 2a2fc26..0521229 100644 --- a/packages/express/ws-server.ts +++ b/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)) } diff --git a/packages/node/locks.ts b/packages/node/locks.ts index 6853795..ffaab20 100644 --- a/packages/node/locks.ts +++ b/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) }