1  import buffers from './testing/test-buffers'
  2  import BufferList from './testing/buffer-list'
  3  import { parse } from '.'
  4  import assert from 'assert'
  5  import { PassThrough } from 'stream'
  6  import { BackendMessage } from './messages'
  7  
  8  const authOkBuffer = buffers.authenticationOk()
  9  const paramStatusBuffer = buffers.parameterStatus('client_encoding', 'UTF8')
 10  const readyForQueryBuffer = buffers.readyForQuery()
 11  const backendKeyDataBuffer = buffers.backendKeyData(1, 2)
 12  const commandCompleteBuffer = buffers.commandComplete('SELECT 3')
 13  const parseCompleteBuffer = buffers.parseComplete()
 14  const bindCompleteBuffer = buffers.bindComplete()
 15  const portalSuspendedBuffer = buffers.portalSuspended()
 16  
 17  const row1 = {
 18    name: 'id',
 19    tableID: 1,
 20    attributeNumber: 2,
 21    dataTypeID: 3,
 22    dataTypeSize: 4,
 23    typeModifier: 5,
 24    formatCode: 0,
 25  }
 26  const oneRowDescBuff = buffers.rowDescription([row1])
 27  row1.name = 'bang'
 28  
 29  const twoRowBuf = buffers.rowDescription([
 30    row1,
 31    {
 32      name: 'whoah',
 33      tableID: 10,
 34      attributeNumber: 11,
 35      dataTypeID: 12,
 36      dataTypeSize: 13,
 37      typeModifier: 14,
 38      formatCode: 0,
 39    },
 40  ])
 41  
 42  const rowWithBigOids = {
 43    name: 'bigoid',
 44    tableID: 3000000001,
 45    attributeNumber: 2,
 46    dataTypeID: 3000000003,
 47    dataTypeSize: 4,
 48    typeModifier: 5,
 49    formatCode: 0,
 50  }
 51  const bigOidDescBuff = buffers.rowDescription([rowWithBigOids])
 52  
 53  const emptyRowFieldBuf = buffers.dataRow([])
 54  
 55  const oneFieldBuf = buffers.dataRow(['test'])
 56  
 57  const expectedAuthenticationOkayMessage = {
 58    name: 'authenticationOk',
 59    length: 8,
 60  }
 61  
 62  const expectedParameterStatusMessage = {
 63    name: 'parameterStatus',
 64    parameterName: 'client_encoding',
 65    parameterValue: 'UTF8',
 66    length: 25,
 67  }
 68  
 69  const expectedBackendKeyDataMessage = {
 70    name: 'backendKeyData',
 71    processID: 1,
 72    secretKey: 2,
 73  }
 74  
 75  const expectedReadyForQueryMessage = {
 76    name: 'readyForQuery',
 77    length: 5,
 78    status: 'I',
 79  }
 80  
 81  const expectedCommandCompleteMessage = {
 82    name: 'commandComplete',
 83    length: 13,
 84    text: 'SELECT 3',
 85  }
 86  const emptyRowDescriptionBuffer = new BufferList()
 87    .addInt16(0) // number of fields
 88    .join(true, 'T')
 89  
 90  const expectedEmptyRowDescriptionMessage = {
 91    name: 'rowDescription',
 92    length: 6,
 93    fieldCount: 0,
 94    fields: [],
 95  }
 96  const expectedOneRowMessage = {
 97    name: 'rowDescription',
 98    length: 27,
 99    fieldCount: 1,
100    fields: [
101      {
102        name: 'id',
103        tableID: 1,
104        columnID: 2,
105        dataTypeID: 3,
106        dataTypeSize: 4,
107        dataTypeModifier: 5,
108        format: 'text',
109      },
110    ],
111  }
112  
113  const expectedTwoRowMessage = {
114    name: 'rowDescription',
115    length: 53,
116    fieldCount: 2,
117    fields: [
118      {
119        name: 'bang',
120        tableID: 1,
121        columnID: 2,
122        dataTypeID: 3,
123        dataTypeSize: 4,
124        dataTypeModifier: 5,
125        format: 'text',
126      },
127      {
128        name: 'whoah',
129        tableID: 10,
130        columnID: 11,
131        dataTypeID: 12,
132        dataTypeSize: 13,
133        dataTypeModifier: 14,
134        format: 'text',
135      },
136    ],
137  }
138  const expectedBigOidMessage = {
139    name: 'rowDescription',
140    length: 31,
141    fieldCount: 1,
142    fields: [
143      {
144        name: 'bigoid',
145        tableID: 3000000001,
146        columnID: 2,
147        dataTypeID: 3000000003,
148        dataTypeSize: 4,
149        dataTypeModifier: 5,
150        format: 'text',
151      },
152    ],
153  }
154  
155  const emptyParameterDescriptionBuffer = new BufferList()
156    .addInt16(0) // number of parameters
157    .join(true, 't')
158  
159  const oneParameterDescBuf = buffers.parameterDescription([1111])
160  
161  const twoParameterDescBuf = buffers.parameterDescription([2222, 3333])
162  
163  const expectedEmptyParameterDescriptionMessage = {
164    name: 'parameterDescription',
165    length: 6,
166    parameterCount: 0,
167    dataTypeIDs: [],
168  }
169  
170  const expectedOneParameterMessage = {
171    name: 'parameterDescription',
172    length: 10,
173    parameterCount: 1,
174    dataTypeIDs: [1111],
175  }
176  
177  const expectedTwoParameterMessage = {
178    name: 'parameterDescription',
179    length: 14,
180    parameterCount: 2,
181    dataTypeIDs: [2222, 3333],
182  }
183  
184  const testForMessage = function (buffer: Buffer, expectedMessage: any) {
185    it('receives and parses ' + expectedMessage.name, async () => {
186      const messages = await parseBuffers([buffer])
187      const [lastMessage] = messages
188  
189      for (const key in expectedMessage) {
190        assert.deepEqual((lastMessage as any)[key], expectedMessage[key])
191      }
192    })
193  }
194  
195  const plainPasswordBuffer = buffers.authenticationCleartextPassword()
196  const md5PasswordBuffer = buffers.authenticationMD5Password()
197  const SASLBuffer = buffers.authenticationSASL()
198  const SASLContinueBuffer = buffers.authenticationSASLContinue()
199  const SASLFinalBuffer = buffers.authenticationSASLFinal()
200  
201  const expectedPlainPasswordMessage = {
202    name: 'authenticationCleartextPassword',
203  }
204  
205  const expectedMD5PasswordMessage = {
206    name: 'authenticationMD5Password',
207    salt: Buffer.from([1, 2, 3, 4]),
208  }
209  
210  const expectedSASLMessage = {
211    name: 'authenticationSASL',
212    mechanisms: ['SCRAM-SHA-256'],
213  }
214  
215  const expectedSASLContinueMessage = {
216    name: 'authenticationSASLContinue',
217    data: 'data',
218  }
219  
220  const expectedSASLFinalMessage = {
221    name: 'authenticationSASLFinal',
222    data: 'data',
223  }
224  
225  const notificationResponseBuffer = buffers.notification(4, 'hi', 'boom')
226  const expectedNotificationResponseMessage = {
227    name: 'notification',
228    processId: 4,
229    channel: 'hi',
230    payload: 'boom',
231  }
232  
233  const parseBuffers = async (buffers: Buffer[]): Promise<BackendMessage[]> => {
234    const stream = new PassThrough()
235    for (const buffer of buffers) {
236      stream.write(buffer)
237    }
238    stream.end()
239    const msgs: BackendMessage[] = []
240    await parse(stream, (msg) => msgs.push(msg))
241    return msgs
242  }
243  
244  describe('PgPacketStream', function () {
245    testForMessage(authOkBuffer, expectedAuthenticationOkayMessage)
246    testForMessage(plainPasswordBuffer, expectedPlainPasswordMessage)
247    testForMessage(md5PasswordBuffer, expectedMD5PasswordMessage)
248    testForMessage(SASLBuffer, expectedSASLMessage)
249    testForMessage(SASLContinueBuffer, expectedSASLContinueMessage)
250  
251    // this exercises a found bug in the parser:
252    // https://github.com/brianc/node-postgres/pull/2210#issuecomment-627626084
253    // and adds a test which is deterministic, rather than relying on network packet chunking
254    const extendedSASLContinueBuffer = Buffer.concat([SASLContinueBuffer, Buffer.from([1, 2, 3, 4])])
255    testForMessage(extendedSASLContinueBuffer, expectedSASLContinueMessage)
256  
257    testForMessage(SASLFinalBuffer, expectedSASLFinalMessage)
258  
259    // this exercises a found bug in the parser:
260    // https://github.com/brianc/node-postgres/pull/2210#issuecomment-627626084
261    // and adds a test which is deterministic, rather than relying on network packet chunking
262    const extendedSASLFinalBuffer = Buffer.concat([SASLFinalBuffer, Buffer.from([1, 2, 4, 5])])
263    testForMessage(extendedSASLFinalBuffer, expectedSASLFinalMessage)
264  
265    testForMessage(paramStatusBuffer, expectedParameterStatusMessage)
266    testForMessage(backendKeyDataBuffer, expectedBackendKeyDataMessage)
267    testForMessage(readyForQueryBuffer, expectedReadyForQueryMessage)
268    testForMessage(commandCompleteBuffer, expectedCommandCompleteMessage)
269    testForMessage(notificationResponseBuffer, expectedNotificationResponseMessage)
270    testForMessage(buffers.emptyQuery(), {
271      name: 'emptyQuery',
272      length: 4,
273    })
274  
275    testForMessage(Buffer.from([0x6e, 0, 0, 0, 4]), {
276      name: 'noData',
277    })
278  
279    describe('rowDescription messages', function () {
280      testForMessage(emptyRowDescriptionBuffer, expectedEmptyRowDescriptionMessage)
281      testForMessage(oneRowDescBuff, expectedOneRowMessage)
282      testForMessage(twoRowBuf, expectedTwoRowMessage)
283      testForMessage(bigOidDescBuff, expectedBigOidMessage)
284    })
285  
286    describe('parameterDescription messages', function () {
287      testForMessage(emptyParameterDescriptionBuffer, expectedEmptyParameterDescriptionMessage)
288      testForMessage(oneParameterDescBuf, expectedOneParameterMessage)
289      testForMessage(twoParameterDescBuf, expectedTwoParameterMessage)
290    })
291  
292    describe('parsing rows', function () {
293      describe('parsing empty row', function () {
294        testForMessage(emptyRowFieldBuf, {
295          name: 'dataRow',
296          fieldCount: 0,
297        })
298      })
299  
300      describe('parsing data row with fields', function () {
301        testForMessage(oneFieldBuf, {
302          name: 'dataRow',
303          fieldCount: 1,
304          fields: ['test'],
305        })
306      })
307    })
308  
309    describe('notice message', function () {
310      // this uses the same logic as error message
311      const buff = buffers.notice([{ type: 'C', value: 'code' }])
312      testForMessage(buff, {
313        name: 'notice',
314        code: 'code',
315      })
316    })
317  
318    testForMessage(buffers.error([]), {
319      name: 'error',
320    })
321  
322    describe('with all the fields', function () {
323      const buffer = buffers.error([
324        {
325          type: 'S',
326          value: 'ERROR',
327        },
328        {
329          type: 'C',
330          value: 'code',
331        },
332        {
333          type: 'M',
334          value: 'message',
335        },
336        {
337          type: 'D',
338          value: 'details',
339        },
340        {
341          type: 'H',
342          value: 'hint',
343        },
344        {
345          type: 'P',
346          value: '100',
347        },
348        {
349          type: 'p',
350          value: '101',
351        },
352        {
353          type: 'q',
354          value: 'query',
355        },
356        {
357          type: 'W',
358          value: 'where',
359        },
360        {
361          type: 'F',
362          value: 'file',
363        },
364        {
365          type: 'L',
366          value: 'line',
367        },
368        {
369          type: 'R',
370          value: 'routine',
371        },
372        {
373          type: 'Z', // ignored
374          value: 'alsdkf',
375        },
376      ])
377  
378      testForMessage(buffer, {
379        name: 'error',
380        severity: 'ERROR',
381        code: 'code',
382        message: 'message',
383        detail: 'details',
384        hint: 'hint',
385        position: '100',
386        internalPosition: '101',
387        internalQuery: 'query',
388        where: 'where',
389        file: 'file',
390        line: 'line',
391        routine: 'routine',
392      })
393    })
394  
395    testForMessage(parseCompleteBuffer, {
396      name: 'parseComplete',
397    })
398  
399    testForMessage(bindCompleteBuffer, {
400      name: 'bindComplete',
401    })
402  
403    testForMessage(bindCompleteBuffer, {
404      name: 'bindComplete',
405    })
406  
407    testForMessage(buffers.closeComplete(), {
408      name: 'closeComplete',
409    })
410  
411    describe('parses portal suspended message', function () {
412      testForMessage(portalSuspendedBuffer, {
413        name: 'portalSuspended',
414      })
415    })
416  
417    describe('parses replication start message', function () {
418      testForMessage(Buffer.from([0x57, 0x00, 0x00, 0x00, 0x04]), {
419        name: 'replicationStart',
420        length: 4,
421      })
422    })
423  
424    describe('copy', () => {
425      testForMessage(buffers.copyIn(0), {
426        name: 'copyInResponse',
427        length: 7,
428        binary: false,
429        columnTypes: [],
430      })
431  
432      testForMessage(buffers.copyIn(2), {
433        name: 'copyInResponse',
434        length: 11,
435        binary: false,
436        columnTypes: [0, 1],
437      })
438  
439      testForMessage(buffers.copyOut(0), {
440        name: 'copyOutResponse',
441        length: 7,
442        binary: false,
443        columnTypes: [],
444      })
445  
446      testForMessage(buffers.copyOut(3), {
447        name: 'copyOutResponse',
448        length: 13,
449        binary: false,
450        columnTypes: [0, 1, 2],
451      })
452  
453      testForMessage(buffers.copyDone(), {
454        name: 'copyDone',
455        length: 4,
456      })
457  
458      testForMessage(buffers.copyData(Buffer.from([5, 6, 7])), {
459        name: 'copyData',
460        length: 7,
461        chunk: Buffer.from([5, 6, 7]),
462      })
463    })
464  
465    // since the data message on a stream can randomly divide the incomming
466    // tcp packets anywhere, we need to make sure we can parse every single
467    // split on a tcp message
468    describe('split buffer, single message parsing', function () {
469      const fullBuffer = buffers.dataRow([null, 'bang', 'zug zug', null, '!'])
470  
471      it('parses when full buffer comes in', async function () {
472        const messages = await parseBuffers([fullBuffer])
473        const message = messages[0] as any
474        assert.equal(message.fields.length, 5)
475        assert.equal(message.fields[0], null)
476        assert.equal(message.fields[1], 'bang')
477        assert.equal(message.fields[2], 'zug zug')
478        assert.equal(message.fields[3], null)
479        assert.equal(message.fields[4], '!')
480      })
481  
482      const testMessageReceivedAfterSplitAt = async function (split: number) {
483        const firstBuffer = Buffer.alloc(fullBuffer.length - split)
484        const secondBuffer = Buffer.alloc(fullBuffer.length - firstBuffer.length)
485        fullBuffer.copy(firstBuffer, 0, 0)
486        fullBuffer.copy(secondBuffer, 0, firstBuffer.length)
487        const messages = await parseBuffers([firstBuffer, secondBuffer])
488        const message = messages[0] as any
489        assert.equal(message.fields.length, 5)
490        assert.equal(message.fields[0], null)
491        assert.equal(message.fields[1], 'bang')
492        assert.equal(message.fields[2], 'zug zug')
493        assert.equal(message.fields[3], null)
494        assert.equal(message.fields[4], '!')
495      }
496  
497      it('parses when split in the middle', function () {
498        return testMessageReceivedAfterSplitAt(6)
499      })
500  
501      it('parses when split at end', function () {
502        return testMessageReceivedAfterSplitAt(2)
503      })
504  
505      it('parses when split at beginning', function () {
506        return Promise.all([
507          testMessageReceivedAfterSplitAt(fullBuffer.length - 2),
508          testMessageReceivedAfterSplitAt(fullBuffer.length - 1),
509          testMessageReceivedAfterSplitAt(fullBuffer.length - 5),
510        ])
511      })
512    })
513  
514    describe('split buffer, multiple message parsing', function () {
515      const dataRowBuffer = buffers.dataRow(['!'])
516      const readyForQueryBuffer = buffers.readyForQuery()
517      const fullBuffer = Buffer.alloc(dataRowBuffer.length + readyForQueryBuffer.length)
518      dataRowBuffer.copy(fullBuffer, 0, 0)
519      readyForQueryBuffer.copy(fullBuffer, dataRowBuffer.length, 0)
520  
521      const verifyMessages = function (messages: any[]) {
522        assert.strictEqual(messages.length, 2)
523        assert.deepEqual(messages[0], {
524          name: 'dataRow',
525          fieldCount: 1,
526          length: 11,
527          fields: ['!'],
528        })
529        assert.equal(messages[0].fields[0], '!')
530        assert.deepEqual(messages[1], {
531          name: 'readyForQuery',
532          length: 5,
533          status: 'I',
534        })
535      }
536      // sanity check
537      it('receives both messages when packet is not split', async function () {
538        const messages = await parseBuffers([fullBuffer])
539        verifyMessages(messages)
540      })
541  
542      const splitAndVerifyTwoMessages = async function (split: number) {
543        const firstBuffer = Buffer.alloc(fullBuffer.length - split)
544        const secondBuffer = Buffer.alloc(fullBuffer.length - firstBuffer.length)
545        fullBuffer.copy(firstBuffer, 0, 0)
546        fullBuffer.copy(secondBuffer, 0, firstBuffer.length)
547        const messages = await parseBuffers([firstBuffer, secondBuffer])
548        verifyMessages(messages)
549      }
550  
551      describe('receives both messages when packet is split', function () {
552        it('in the middle', function () {
553          return splitAndVerifyTwoMessages(11)
554        })
555        it('at the front', function () {
556          return Promise.all([
557            splitAndVerifyTwoMessages(fullBuffer.length - 1),
558            splitAndVerifyTwoMessages(fullBuffer.length - 4),
559            splitAndVerifyTwoMessages(fullBuffer.length - 6),
560          ])
561        })
562  
563        it('at the end', function () {
564          return Promise.all([splitAndVerifyTwoMessages(8), splitAndVerifyTwoMessages(1)])
565        })
566      })
567    })
568  })