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) }