1  /*!
  2   * raw-body
  3   * Copyright(c) 2013-2014 Jonathan Ong
  4   * Copyright(c) 2014-2022 Douglas Christopher Wilson
  5   * MIT Licensed
  6   */
  7  
  8  'use strict'
  9  
 10  /**
 11   * Module dependencies.
 12   * @private
 13   */
 14  
 15  var asyncHooks = tryRequireAsyncHooks()
 16  var bytes = require('bytes')
 17  var createError = require('http-errors')
 18  var iconv = require('iconv-lite')
 19  var unpipe = require('unpipe')
 20  
 21  /**
 22   * Module exports.
 23   * @public
 24   */
 25  
 26  module.exports = getRawBody
 27  
 28  /**
 29   * Module variables.
 30   * @private
 31   */
 32  
 33  var ICONV_ENCODING_MESSAGE_REGEXP = /^Encoding not recognized: /
 34  
 35  /**
 36   * Get the decoder for a given encoding.
 37   *
 38   * @param {string} encoding
 39   * @private
 40   */
 41  
 42  function getDecoder (encoding) {
 43    if (!encoding) return null
 44  
 45    try {
 46      return iconv.getDecoder(encoding)
 47    } catch (e) {
 48      // error getting decoder
 49      if (!ICONV_ENCODING_MESSAGE_REGEXP.test(e.message)) throw e
 50  
 51      // the encoding was not found
 52      throw createError(415, 'specified encoding unsupported', {
 53        encoding: encoding,
 54        type: 'encoding.unsupported'
 55      })
 56    }
 57  }
 58  
 59  /**
 60   * Get the raw body of a stream (typically HTTP).
 61   *
 62   * @param {object} stream
 63   * @param {object|string|function} [options]
 64   * @param {function} [callback]
 65   * @public
 66   */
 67  
 68  function getRawBody (stream, options, callback) {
 69    var done = callback
 70    var opts = options || {}
 71  
 72    // light validation
 73    if (stream === undefined) {
 74      throw new TypeError('argument stream is required')
 75    } else if (typeof stream !== 'object' || stream === null || typeof stream.on !== 'function') {
 76      throw new TypeError('argument stream must be a stream')
 77    }
 78  
 79    if (options === true || typeof options === 'string') {
 80      // short cut for encoding
 81      opts = {
 82        encoding: options
 83      }
 84    }
 85  
 86    if (typeof options === 'function') {
 87      done = options
 88      opts = {}
 89    }
 90  
 91    // validate callback is a function, if provided
 92    if (done !== undefined && typeof done !== 'function') {
 93      throw new TypeError('argument callback must be a function')
 94    }
 95  
 96    // require the callback without promises
 97    if (!done && !global.Promise) {
 98      throw new TypeError('argument callback is required')
 99    }
100  
101    // get encoding
102    var encoding = opts.encoding !== true
103      ? opts.encoding
104      : 'utf-8'
105  
106    // convert the limit to an integer
107    var limit = bytes.parse(opts.limit)
108  
109    // convert the expected length to an integer
110    var length = opts.length != null && !isNaN(opts.length)
111      ? parseInt(opts.length, 10)
112      : null
113  
114    if (done) {
115      // classic callback style
116      return readStream(stream, encoding, length, limit, wrap(done))
117    }
118  
119    return new Promise(function executor (resolve, reject) {
120      readStream(stream, encoding, length, limit, function onRead (err, buf) {
121        if (err) return reject(err)
122        resolve(buf)
123      })
124    })
125  }
126  
127  /**
128   * Halt a stream.
129   *
130   * @param {Object} stream
131   * @private
132   */
133  
134  function halt (stream) {
135    // unpipe everything from the stream
136    unpipe(stream)
137  
138    // pause stream
139    if (typeof stream.pause === 'function') {
140      stream.pause()
141    }
142  }
143  
144  /**
145   * Read the data from the stream.
146   *
147   * @param {object} stream
148   * @param {string} encoding
149   * @param {number} length
150   * @param {number} limit
151   * @param {function} callback
152   * @public
153   */
154  
155  function readStream (stream, encoding, length, limit, callback) {
156    var complete = false
157    var sync = true
158  
159    // check the length and limit options.
160    // note: we intentionally leave the stream paused,
161    // so users should handle the stream themselves.
162    if (limit !== null && length !== null && length > limit) {
163      return done(createError(413, 'request entity too large', {
164        expected: length,
165        length: length,
166        limit: limit,
167        type: 'entity.too.large'
168      }))
169    }
170  
171    // streams1: assert request encoding is buffer.
172    // streams2+: assert the stream encoding is buffer.
173    //   stream._decoder: streams1
174    //   state.encoding: streams2
175    //   state.decoder: streams2, specifically < 0.10.6
176    var state = stream._readableState
177    if (stream._decoder || (state && (state.encoding || state.decoder))) {
178      // developer error
179      return done(createError(500, 'stream encoding should not be set', {
180        type: 'stream.encoding.set'
181      }))
182    }
183  
184    if (typeof stream.readable !== 'undefined' && !stream.readable) {
185      return done(createError(500, 'stream is not readable', {
186        type: 'stream.not.readable'
187      }))
188    }
189  
190    var received = 0
191    var decoder
192  
193    try {
194      decoder = getDecoder(encoding)
195    } catch (err) {
196      return done(err)
197    }
198  
199    var buffer = decoder
200      ? ''
201      : []
202  
203    // attach listeners
204    stream.on('aborted', onAborted)
205    stream.on('close', cleanup)
206    stream.on('data', onData)
207    stream.on('end', onEnd)
208    stream.on('error', onEnd)
209  
210    // mark sync section complete
211    sync = false
212  
213    function done () {
214      var args = new Array(arguments.length)
215  
216      // copy arguments
217      for (var i = 0; i < args.length; i++) {
218        args[i] = arguments[i]
219      }
220  
221      // mark complete
222      complete = true
223  
224      if (sync) {
225        process.nextTick(invokeCallback)
226      } else {
227        invokeCallback()
228      }
229  
230      function invokeCallback () {
231        cleanup()
232  
233        if (args[0]) {
234          // halt the stream on error
235          halt(stream)
236        }
237  
238        callback.apply(null, args)
239      }
240    }
241  
242    function onAborted () {
243      if (complete) return
244  
245      done(createError(400, 'request aborted', {
246        code: 'ECONNABORTED',
247        expected: length,
248        length: length,
249        received: received,
250        type: 'request.aborted'
251      }))
252    }
253  
254    function onData (chunk) {
255      if (complete) return
256  
257      received += chunk.length
258  
259      if (limit !== null && received > limit) {
260        done(createError(413, 'request entity too large', {
261          limit: limit,
262          received: received,
263          type: 'entity.too.large'
264        }))
265      } else if (decoder) {
266        buffer += decoder.write(chunk)
267      } else {
268        buffer.push(chunk)
269      }
270    }
271  
272    function onEnd (err) {
273      if (complete) return
274      if (err) return done(err)
275  
276      if (length !== null && received !== length) {
277        done(createError(400, 'request size did not match content length', {
278          expected: length,
279          length: length,
280          received: received,
281          type: 'request.size.invalid'
282        }))
283      } else {
284        var string = decoder
285          ? buffer + (decoder.end() || '')
286          : Buffer.concat(buffer)
287        done(null, string)
288      }
289    }
290  
291    function cleanup () {
292      buffer = null
293  
294      stream.removeListener('aborted', onAborted)
295      stream.removeListener('data', onData)
296      stream.removeListener('end', onEnd)
297      stream.removeListener('error', onEnd)
298      stream.removeListener('close', cleanup)
299    }
300  }
301  
302  /**
303   * Try to require async_hooks
304   * @private
305   */
306  
307  function tryRequireAsyncHooks () {
308    try {
309      return require('async_hooks')
310    } catch (e) {
311      return {}
312    }
313  }
314  
315  /**
316   * Wrap function with async resource, if possible.
317   * AsyncResource.bind static method backported.
318   * @private
319   */
320  
321  function wrap (fn) {
322    var res
323  
324    // create anonymous resource
325    if (asyncHooks.AsyncResource) {
326      res = new asyncHooks.AsyncResource(fn.name || 'bound-anonymous-fn')
327    }
328  
329    // incompatible node.js
330    if (!res || !res.runInAsyncScope) {
331      return fn
332    }
333  
334    // return bound function
335    return res.runInAsyncScope.bind(res, fn, null)
336  }