actions/bcr-pr-reviewer/node_modules/undici/lib/websocket/receiver.js - continuous-integration - Git at Google

 'use strict'

 const { Writable } = require('stream')
 const diagnosticsChannel = require('diagnostics_channel')
 const { parserStates, opcodes, states, emptyBuffer } = require('./constants')
 const { kReadyState, kSentClose, kResponse, kReceivedClose } = require('./symbols')
 const { isValidStatusCode, failWebsocketConnection, websocketMessageReceived } = require('./util')
 const { WebsocketFrameSend } = require('./frame')

 // This code was influenced by ws released under the MIT license.
 // Copyright (c) 2011 Einar Otto Stangvik <einaros@gmail.com>
 // Copyright (c) 2013 Arnout Kazemier and contributors
 // Copyright (c) 2016 Luigi Pinca and contributors

 const channels = {}
 channels.ping = diagnosticsChannel.channel('undici:websocket:ping')
 channels.pong = diagnosticsChannel.channel('undici:websocket:pong')

 class ByteParser extends Writable {
   #buffers = []
   #byteOffset = 0

   #state = parserStates.INFO

   #info = {}
   #fragments = []

   constructor (ws) {
     super()

     this.ws = ws
   }

   /**
    * @param {Buffer} chunk
    * @param {() => void} callback
    */
   _write (chunk, _, callback) {
     this.#buffers.push(chunk)
     this.#byteOffset += chunk.length

     this.run(callback)
   }

   /**
    * Runs whenever a new chunk is received.
    * Callback is called whenever there are no more chunks buffering,
    * or not enough bytes are buffered to parse.
    */
   run (callback) {
     while (true) {
       if (this.#state === parserStates.INFO) {
         // If there aren't enough bytes to parse the payload length, etc.
         if (this.#byteOffset < 2) {
           return callback()
         }

         const buffer = this.consume(2)

         this.#info.fin = (buffer[0] & 0x80) !== 0
         this.#info.opcode = buffer[0] & 0x0F

         // If we receive a fragmented message, we use the type of the first
         // frame to parse the full message as binary/text, when it's terminated
         this.#info.originalOpcode ??= this.#info.opcode

         this.#info.fragmented = !this.#info.fin && this.#info.opcode !== opcodes.CONTINUATION

         if (this.#info.fragmented && this.#info.opcode !== opcodes.BINARY && this.#info.opcode !== opcodes.TEXT) {
           // Only text and binary frames can be fragmented
           failWebsocketConnection(this.ws, 'Invalid frame type was fragmented.')
           return
         }

         const payloadLength = buffer[1] & 0x7F

         if (payloadLength <= 125) {
           this.#info.payloadLength = payloadLength
           this.#state = parserStates.READ_DATA
         } else if (payloadLength === 126) {
           this.#state = parserStates.PAYLOADLENGTH_16
         } else if (payloadLength === 127) {
           this.#state = parserStates.PAYLOADLENGTH_64
         }

         if (this.#info.fragmented && payloadLength > 125) {
           // A fragmented frame can't be fragmented itself
           failWebsocketConnection(this.ws, 'Fragmented frame exceeded 125 bytes.')
           return
         } else if (
           (this.#info.opcode === opcodes.PING ||
             this.#info.opcode === opcodes.PONG ||
             this.#info.opcode === opcodes.CLOSE) &&
           payloadLength > 125
         ) {
           // Control frames can have a payload length of 125 bytes MAX
           failWebsocketConnection(this.ws, 'Payload length for control frame exceeded 125 bytes.')
           return
         } else if (this.#info.opcode === opcodes.CLOSE) {
           if (payloadLength === 1) {
             failWebsocketConnection(this.ws, 'Received close frame with a 1-byte body.')
             return
           }

           const body = this.consume(payloadLength)

           this.#info.closeInfo = this.parseCloseBody(false, body)

           if (!this.ws[kSentClose]) {
             // If an endpoint receives a Close frame and did not previously send a
             // Close frame, the endpoint MUST send a Close frame in response.  (When
             // sending a Close frame in response, the endpoint typically echos the
             // status code it received.)
             const body = Buffer.allocUnsafe(2)
             body.writeUInt16BE(this.#info.closeInfo.code, 0)
             const closeFrame = new WebsocketFrameSend(body)

             this.ws[kResponse].socket.write(
               closeFrame.createFrame(opcodes.CLOSE),
               (err) => {
                 if (!err) {
                   this.ws[kSentClose] = true
                 }
               }
             )
           }

           // Upon either sending or receiving a Close control frame, it is said
           // that _The WebSocket Closing Handshake is Started_ and that the
           // WebSocket connection is in the CLOSING state.
           this.ws[kReadyState] = states.CLOSING
           this.ws[kReceivedClose] = true

           this.end()

           return
         } else if (this.#info.opcode === opcodes.PING) {
           // Upon receipt of a Ping frame, an endpoint MUST send a Pong frame in
           // response, unless it already received a Close frame.
           // A Pong frame sent in response to a Ping frame must have identical
           // "Application data"

           const body = this.consume(payloadLength)

           if (!this.ws[kReceivedClose]) {
             const frame = new WebsocketFrameSend(body)

             this.ws[kResponse].socket.write(frame.createFrame(opcodes.PONG))

             if (channels.ping.hasSubscribers) {
               channels.ping.publish({
                 payload: body
               })
             }
           }

           this.#state = parserStates.INFO

           if (this.#byteOffset > 0) {
             continue
           } else {
             callback()
             return
           }
         } else if (this.#info.opcode === opcodes.PONG) {
           // A Pong frame MAY be sent unsolicited.  This serves as a
           // unidirectional heartbeat.  A response to an unsolicited Pong frame is
           // not expected.

           const body = this.consume(payloadLength)

           if (channels.pong.hasSubscribers) {
             channels.pong.publish({
               payload: body
             })
           }

           if (this.#byteOffset > 0) {
             continue
           } else {
             callback()
             return
           }
         }
       } else if (this.#state === parserStates.PAYLOADLENGTH_16) {
         if (this.#byteOffset < 2) {
           return callback()
         }

         const buffer = this.consume(2)

         this.#info.payloadLength = buffer.readUInt16BE(0)
         this.#state = parserStates.READ_DATA
       } else if (this.#state === parserStates.PAYLOADLENGTH_64) {
         if (this.#byteOffset < 8) {
           return callback()
         }

         const buffer = this.consume(8)
         const upper = buffer.readUInt32BE(0)

         // 2^31 is the maxinimum bytes an arraybuffer can contain
         // on 32-bit systems. Although, on 64-bit systems, this is
         // 2^53-1 bytes.
         // https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Errors/Invalid_array_length
         // https://source.chromium.org/chromium/chromium/src/+/main:v8/src/common/globals.h;drc=1946212ac0100668f14eb9e2843bdd846e510a1e;bpv=1;bpt=1;l=1275
         // https://source.chromium.org/chromium/chromium/src/+/main:v8/src/objects/js-array-buffer.h;l=34;drc=1946212ac0100668f14eb9e2843bdd846e510a1e
         if (upper > 2 ** 31 - 1) {
           failWebsocketConnection(this.ws, 'Received payload length > 2^31 bytes.')
           return
         }

         const lower = buffer.readUInt32BE(4)

         this.#info.payloadLength = (upper << 8) + lower
         this.#state = parserStates.READ_DATA
       } else if (this.#state === parserStates.READ_DATA) {
         if (this.#byteOffset < this.#info.payloadLength) {
           // If there is still more data in this chunk that needs to be read
           return callback()
         } else if (this.#byteOffset >= this.#info.payloadLength) {
           // If the server sent multiple frames in a single chunk

           const body = this.consume(this.#info.payloadLength)

           this.#fragments.push(body)

           // If the frame is unfragmented, or a fragmented frame was terminated,
           // a message was received
           if (!this.#info.fragmented || (this.#info.fin && this.#info.opcode === opcodes.CONTINUATION)) {
             const fullMessage = Buffer.concat(this.#fragments)

             websocketMessageReceived(this.ws, this.#info.originalOpcode, fullMessage)

             this.#info = {}
             this.#fragments.length = 0
           }

           this.#state = parserStates.INFO
         }
       }

       if (this.#byteOffset > 0) {
         continue
       } else {
         callback()
         break
       }
     }
   }

   /**
    * Take n bytes from the buffered Buffers
    * @param {number} n
    * @returns {Buffer|null}
    */
   consume (n) {
     if (n > this.#byteOffset) {
       return null
     } else if (n === 0) {
       return emptyBuffer
     }

     if (this.#buffers[0].length === n) {
       this.#byteOffset -= this.#buffers[0].length
       return this.#buffers.shift()
     }

     const buffer = Buffer.allocUnsafe(n)
     let offset = 0

     while (offset !== n) {
       const next = this.#buffers[0]
       const { length } = next

       if (length + offset === n) {
         buffer.set(this.#buffers.shift(), offset)
         break
       } else if (length + offset > n) {
         buffer.set(next.subarray(0, n - offset), offset)
         this.#buffers[0] = next.subarray(n - offset)
         break
       } else {
         buffer.set(this.#buffers.shift(), offset)
         offset += next.length
       }
     }

     this.#byteOffset -= n

     return buffer
   }

   parseCloseBody (onlyCode, data) {
     // https://datatracker.ietf.org/doc/html/rfc6455#section-7.1.5
     /** @type {number|undefined} */
     let code

     if (data.length >= 2) {
       // _The WebSocket Connection Close Code_ is
       // defined as the status code (Section 7.4) contained in the first Close
       // control frame received by the application
       code = data.readUInt16BE(0)
     }

     if (onlyCode) {
       if (!isValidStatusCode(code)) {
         return null
       }

       return { code }
     }

     // https://datatracker.ietf.org/doc/html/rfc6455#section-7.1.6
     /** @type {Buffer} */
     let reason = data.subarray(2)

     // Remove BOM
     if (reason[0] === 0xEF && reason[1] === 0xBB && reason[2] === 0xBF) {
       reason = reason.subarray(3)
     }

     if (code !== undefined && !isValidStatusCode(code)) {
       return null
     }

     try {
       // TODO: optimize this
       reason = new TextDecoder('utf-8', { fatal: true }).decode(reason)
     } catch {
       return null
     }

     return { code, reason }
   }

   get closingInfo () {
     return this.#info.closeInfo
   }
 }

 module.exports = {
   ByteParser
 }
	'use strict'

	const { Writable } = require('stream')
	const diagnosticsChannel = require('diagnostics_channel')
	const { parserStates, opcodes, states, emptyBuffer } = require('./constants')
	const { kReadyState, kSentClose, kResponse, kReceivedClose } = require('./symbols')
	const { isValidStatusCode, failWebsocketConnection, websocketMessageReceived } = require('./util')
	const { WebsocketFrameSend } = require('./frame')

	// This code was influenced by ws released under the MIT license.
	// Copyright (c) 2011 Einar Otto Stangvik <einaros@gmail.com>
	// Copyright (c) 2013 Arnout Kazemier and contributors
	// Copyright (c) 2016 Luigi Pinca and contributors

	const channels = {}
	channels.ping = diagnosticsChannel.channel('undici:websocket:ping')
	channels.pong = diagnosticsChannel.channel('undici:websocket:pong')

	class ByteParser extends Writable {
	#buffers = []
	#byteOffset = 0

	#state = parserStates.INFO

	#info = {}
	#fragments = []

	constructor (ws) {
	super()

	this.ws = ws
	}

	/**
	* @param {Buffer} chunk
	* @param {() => void} callback
	*/
	_write (chunk, _, callback) {
	this.#buffers.push(chunk)
	this.#byteOffset += chunk.length

	this.run(callback)
	}

	/**
	* Runs whenever a new chunk is received.
	* Callback is called whenever there are no more chunks buffering,
	* or not enough bytes are buffered to parse.
	*/
	run (callback) {
	while (true) {
	if (this.#state === parserStates.INFO) {
	// If there aren't enough bytes to parse the payload length, etc.
	if (this.#byteOffset < 2) {
	return callback()
	}

	const buffer = this.consume(2)

	this.#info.fin = (buffer[0] & 0x80) !== 0
	this.#info.opcode = buffer[0] & 0x0F

	// If we receive a fragmented message, we use the type of the first
	// frame to parse the full message as binary/text, when it's terminated
	this.#info.originalOpcode ??= this.#info.opcode

	this.#info.fragmented = !this.#info.fin && this.#info.opcode !== opcodes.CONTINUATION

	if (this.#info.fragmented && this.#info.opcode !== opcodes.BINARY && this.#info.opcode !== opcodes.TEXT) {
	// Only text and binary frames can be fragmented
	failWebsocketConnection(this.ws, 'Invalid frame type was fragmented.')
	return
	}

	const payloadLength = buffer[1] & 0x7F

	if (payloadLength <= 125) {
	this.#info.payloadLength = payloadLength
	this.#state = parserStates.READ_DATA
	} else if (payloadLength === 126) {
	this.#state = parserStates.PAYLOADLENGTH_16
	} else if (payloadLength === 127) {
	this.#state = parserStates.PAYLOADLENGTH_64
	}

	if (this.#info.fragmented && payloadLength > 125) {
	// A fragmented frame can't be fragmented itself
	failWebsocketConnection(this.ws, 'Fragmented frame exceeded 125 bytes.')
	return
	} else if (
	(this.#info.opcode === opcodes.PING \|\|
	this.#info.opcode === opcodes.PONG \|\|
	this.#info.opcode === opcodes.CLOSE) &&
	payloadLength > 125
	) {
	// Control frames can have a payload length of 125 bytes MAX
	failWebsocketConnection(this.ws, 'Payload length for control frame exceeded 125 bytes.')
	return
	} else if (this.#info.opcode === opcodes.CLOSE) {
	if (payloadLength === 1) {
	failWebsocketConnection(this.ws, 'Received close frame with a 1-byte body.')
	return
	}

	const body = this.consume(payloadLength)

	this.#info.closeInfo = this.parseCloseBody(false, body)

	if (!this.ws[kSentClose]) {
	// If an endpoint receives a Close frame and did not previously send a
	// Close frame, the endpoint MUST send a Close frame in response. (When
	// sending a Close frame in response, the endpoint typically echos the
	// status code it received.)
	const body = Buffer.allocUnsafe(2)
	body.writeUInt16BE(this.#info.closeInfo.code, 0)
	const closeFrame = new WebsocketFrameSend(body)

	this.ws[kResponse].socket.write(
	closeFrame.createFrame(opcodes.CLOSE),
	(err) => {
	if (!err) {
	this.ws[kSentClose] = true
	}
	}
	)
	}

	// Upon either sending or receiving a Close control frame, it is said
	// that _The WebSocket Closing Handshake is Started_ and that the
	// WebSocket connection is in the CLOSING state.
	this.ws[kReadyState] = states.CLOSING
	this.ws[kReceivedClose] = true

	this.end()

	return
	} else if (this.#info.opcode === opcodes.PING) {
	// Upon receipt of a Ping frame, an endpoint MUST send a Pong frame in
	// response, unless it already received a Close frame.
	// A Pong frame sent in response to a Ping frame must have identical
	// "Application data"

	const body = this.consume(payloadLength)

	if (!this.ws[kReceivedClose]) {
	const frame = new WebsocketFrameSend(body)

	this.ws[kResponse].socket.write(frame.createFrame(opcodes.PONG))

	if (channels.ping.hasSubscribers) {
	channels.ping.publish({
	payload: body
	})
	}
	}

	this.#state = parserStates.INFO

	if (this.#byteOffset > 0) {
	continue
	} else {
	callback()
	return
	}
	} else if (this.#info.opcode === opcodes.PONG) {
	// A Pong frame MAY be sent unsolicited. This serves as a
	// unidirectional heartbeat. A response to an unsolicited Pong frame is
	// not expected.

	const body = this.consume(payloadLength)

	if (channels.pong.hasSubscribers) {
	channels.pong.publish({
	payload: body
	})
	}

	if (this.#byteOffset > 0) {
	continue
	} else {
	callback()
	return
	}
	}
	} else if (this.#state === parserStates.PAYLOADLENGTH_16) {
	if (this.#byteOffset < 2) {
	return callback()
	}

	const buffer = this.consume(2)

	this.#info.payloadLength = buffer.readUInt16BE(0)
	this.#state = parserStates.READ_DATA
	} else if (this.#state === parserStates.PAYLOADLENGTH_64) {
	if (this.#byteOffset < 8) {
	return callback()
	}

	const buffer = this.consume(8)
	const upper = buffer.readUInt32BE(0)

	// 2^31 is the maxinimum bytes an arraybuffer can contain
	// on 32-bit systems. Although, on 64-bit systems, this is
	// 2^53-1 bytes.
	// https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Errors/Invalid_array_length
	// https://source.chromium.org/chromium/chromium/src/+/main:v8/src/common/globals.h;drc=1946212ac0100668f14eb9e2843bdd846e510a1e;bpv=1;bpt=1;l=1275
	// https://source.chromium.org/chromium/chromium/src/+/main:v8/src/objects/js-array-buffer.h;l=34;drc=1946212ac0100668f14eb9e2843bdd846e510a1e
	if (upper > 2 ** 31 - 1) {
	failWebsocketConnection(this.ws, 'Received payload length > 2^31 bytes.')
	return
	}

	const lower = buffer.readUInt32BE(4)

	this.#info.payloadLength = (upper << 8) + lower
	this.#state = parserStates.READ_DATA
	} else if (this.#state === parserStates.READ_DATA) {
	if (this.#byteOffset < this.#info.payloadLength) {
	// If there is still more data in this chunk that needs to be read
	return callback()
	} else if (this.#byteOffset >= this.#info.payloadLength) {
	// If the server sent multiple frames in a single chunk

	const body = this.consume(this.#info.payloadLength)

	this.#fragments.push(body)

	// If the frame is unfragmented, or a fragmented frame was terminated,
	// a message was received
	if (!this.#info.fragmented \|\| (this.#info.fin && this.#info.opcode === opcodes.CONTINUATION)) {
	const fullMessage = Buffer.concat(this.#fragments)

	websocketMessageReceived(this.ws, this.#info.originalOpcode, fullMessage)

	this.#info = {}
	this.#fragments.length = 0
	}

	this.#state = parserStates.INFO
	}
	}

	if (this.#byteOffset > 0) {
	continue
	} else {
	callback()
	break
	}
	}
	}

	/**
	* Take n bytes from the buffered Buffers
	* @param {number} n
	* @returns {Buffer\|null}
	*/
	consume (n) {
	if (n > this.#byteOffset) {
	return null
	} else if (n === 0) {
	return emptyBuffer
	}

	if (this.#buffers[0].length === n) {
	this.#byteOffset -= this.#buffers[0].length
	return this.#buffers.shift()
	}

	const buffer = Buffer.allocUnsafe(n)
	let offset = 0

	while (offset !== n) {
	const next = this.#buffers[0]
	const { length } = next

	if (length + offset === n) {
	buffer.set(this.#buffers.shift(), offset)
	break
	} else if (length + offset > n) {
	buffer.set(next.subarray(0, n - offset), offset)
	this.#buffers[0] = next.subarray(n - offset)
	break
	} else {
	buffer.set(this.#buffers.shift(), offset)
	offset += next.length
	}
	}

	this.#byteOffset -= n

	return buffer
	}

	parseCloseBody (onlyCode, data) {
	// https://datatracker.ietf.org/doc/html/rfc6455#section-7.1.5
	/** @type {number\|undefined} */
	let code

	if (data.length >= 2) {
	// _The WebSocket Connection Close Code_ is
	// defined as the status code (Section 7.4) contained in the first Close
	// control frame received by the application
	code = data.readUInt16BE(0)
	}

	if (onlyCode) {
	if (!isValidStatusCode(code)) {
	return null
	}

	return { code }
	}

	// https://datatracker.ietf.org/doc/html/rfc6455#section-7.1.6
	/** @type {Buffer} */
	let reason = data.subarray(2)

	// Remove BOM
	if (reason[0] === 0xEF && reason[1] === 0xBB && reason[2] === 0xBF) {
	reason = reason.subarray(3)
	}

	if (code !== undefined && !isValidStatusCode(code)) {
	return null
	}

	try {
	// TODO: optimize this
	reason = new TextDecoder('utf-8', { fatal: true }).decode(reason)
	} catch {
	return null
	}

	return { code, reason }
	}

	get closingInfo () {
	return this.#info.closeInfo
	}
	}

	module.exports = {
	ByteParser
	}