From 44ba4f7acbac681d773eb010a0c339bb20bf4ea4 Mon Sep 17 00:00:00 2001 From: Leng Niu <344562942+vtxe02023274@users.noreply.github.com> Date: Sun, 1 Nov 2026 09:44:27 +0800 Subject: [PATCH] fix(pg-protocol): fail the stream when the parser throws on an unsupported message --- .../pg-protocol/src/inbound-parser.test.ts | 26 +++++++++++++++ packages/pg-protocol/src/index.ts | 27 ++++++++++++++- .../test/integration/gh-issues/3842-tests.js | 33 +++++++++++++++++++ 3 files changed, 85 insertions(+), 1 deletion(-) create mode 100644 packages/pg/test/integration/gh-issues/3842-tests.js diff --git a/packages/pg-protocol/src/inbound-parser.test.ts b/packages/pg-protocol/src/inbound-parser.test.ts index 8687194c3..16ba715de 100644 --- a/packages/pg-protocol/src/inbound-parser.test.ts +++ b/packages/pg-protocol/src/inbound-parser.test.ts @@ -582,4 +582,30 @@ describe('PgPacketStream', function () { parser.parse(oneFieldBuf, () => {}) assert.strictEqual((parser as any).reader.buffer.byteLength, 0) }) + + it('fails the stream instead of throwing when the parser throws', async function () { + const stream = new PassThrough() + const msgs: BackendMessage[] = [] + const errors: Error[] = [] + stream.on('error', (err) => errors.push(err)) + parse(stream, (msg) => msgs.push(msg)) + // AuthenticationGSS (code 7) is not supported by the parser + const gss = new BufferList().addInt32(7).join(true, 'R') + stream.write(Buffer.concat([authOkBuffer, gss])) + await new Promise((resolve) => stream.once('close', resolve)) + assert.strictEqual(msgs.length, 1) + assert.strictEqual(errors.length, 1) + assert.match(errors[0].message, /Unknown authenticationOk message type 7/) + }) + + it('does not swallow errors thrown by the callback', function () { + const stream = new PassThrough() + const errors: Error[] = [] + stream.on('error', (err) => errors.push(err)) + parse(stream, () => { + throw new Error('boom') + }) + assert.throws(() => stream.emit('data', authOkBuffer), /boom/) + assert.strictEqual(errors.length, 0) + }) }) diff --git a/packages/pg-protocol/src/index.ts b/packages/pg-protocol/src/index.ts index 703ff2e49..fc3e88b35 100644 --- a/packages/pg-protocol/src/index.ts +++ b/packages/pg-protocol/src/index.ts @@ -1,10 +1,35 @@ +import { Readable } from 'stream' import { DatabaseError } from './messages' import { serialize } from './serializer' import { Parser, MessageCallback } from './parser' export function parse(stream: NodeJS.ReadableStream, callback: MessageCallback): Promise { const parser = new Parser() - stream.on('data', (buffer: Buffer) => parser.parse(buffer, callback)) + stream.on('data', (buffer: Buffer) => { + // Errors thrown by the callback are the caller's business and must keep propagating. + // Errors thrown by the parser mean the server sent something we cannot handle: fail + // the stream instead of letting the exception escape the 'data' listener. + let callbackThrew = false + try { + parser.parse(buffer, (message) => { + try { + callback(message) + } catch (err) { + callbackThrew = true + throw err + } + }) + } catch (err) { + if (callbackThrew) { + throw err + } + if (typeof (stream as Readable).destroy === 'function') { + ;(stream as Readable).destroy(err as Error) + } else { + stream.emit('error', err) + } + } + }) return new Promise((resolve) => stream.on('end', () => resolve())) } diff --git a/packages/pg/test/integration/gh-issues/3842-tests.js b/packages/pg/test/integration/gh-issues/3842-tests.js new file mode 100644 index 000000000..b8db44e76 --- /dev/null +++ b/packages/pg/test/integration/gh-issues/3842-tests.js @@ -0,0 +1,33 @@ +'use strict' +const helper = require('./../test-helper') +const assert = require('assert') +const net = require('net') +const pg = require('../../../lib') + +const suite = new helper.Suite() + +// a backend that answers the startup message with an authentication request +// pg-protocol does not know how to parse (code 7 is AuthenticationGSS) +suite.test('unsupported authentication request rejects connect', (done) => { + const server = net.createServer((socket) => { + socket.once('data', () => { + const message = Buffer.alloc(9) + message.write('R', 0) + message.writeInt32BE(8, 1) + message.writeInt32BE(7, 5) + socket.write(message) + }) + socket.on('error', () => {}) + socket.on('close', () => server.close()) + }) + server.listen(0, '127.0.0.1', () => { + const port = server.address().port + const client = new pg.Client({ host: '127.0.0.1', port, user: 'app', database: 'app' }) + client.on('error', () => {}) + client.connect((err) => { + assert(err instanceof Error) + assert(/Unknown authenticationOk message type 7/.test(err.message)) + done() + }) + }) +})