Add support for maxResponseSize and maxCompressedResponseSize

docs/basic-config.asciidoc

@@ -259,6 +259,14 @@ _Default:_ `false`
 |`string` - If configured, verify that the fingerprint of the CA certificate that has signed the certificate of the server matches the supplied fingerprint. Only accepts SHA256 digest fingerprints. +
 _Default:_ `null`
 
+|`maxResponseSize`
+|`number` - When configured, it verifies that the uncompressed response size is lower than the configured number, if it's higher it will abort the request. It cannot be higher than buffer.constants.MAX_STRING_LENTGH +
+_Default:_ `null`
+
+|`maxCompressedResponseSize`
+|`number` - When configured, it verifies that the compressed response size is lower than the configured number, if it's higher it will abort the request. It cannot be higher than buffer.constants.MAX_LENTGH +
+_Default:_ `null`
+
 |===
 
 [discrete]

docs/connecting.asciidoc

@@ -418,6 +418,15 @@ _Default:_ `null`
 |`context`
 |`any` - Custom object per request. _(you can use it to pass data to the clients events)_ +
 _Default:_ `null`
+
+|`maxResponseSize`
+|`number` - When configured, it verifies that the uncompressed response size is lower than the configured number, if it's higher it will abort the request. It cannot be higher than buffer.constants.MAX_STRING_LENTGH +
+_Default:_ `null`
+
+|`maxCompressedResponseSize`
+|`number` - When configured, it verifies that the compressed response size is lower than the configured number, if it's higher it will abort the request. It cannot be higher than buffer.constants.MAX_LENTGH +
+_Default:_ `null`
+
 |===
 
 [discrete]

index.d.ts

@@ -119,6 +119,8 @@ interface ClientOptions {
   };
   disablePrototypePoisoningProtection?: boolean | 'proto' | 'constructor';
   caFingerprint?: string;
+  maxResponseSize?: number;
+  maxCompressedResponseSize?: number;
 }
 
 declare class Client {

index.js

@@ -21,6 +21,7 @@
 
 const { EventEmitter } = require('events')
 const { URL } = require('url')
+const buffer = require('buffer')
 const debug = require('debug')('elasticsearch')
 const Transport = require('./lib/Transport')
 const Connection = require('./lib/Connection')
@@ -114,9 +115,19 @@ class Client extends ESAPI {
         context: null,
         proxy: null,
         enableMetaHeader: true,
-        disablePrototypePoisoningProtection: false
+        disablePrototypePoisoningProtection: false,
+        maxResponseSize: null,
+        maxCompressedResponseSize: null
       }, opts)
 
+    if (options.maxResponseSize !== null && options.maxResponseSize > buffer.constants.MAX_STRING_LENGTH) {
+      throw new ConfigurationError(`The maxResponseSize cannot be bigger than ${buffer.constants.MAX_STRING_LENGTH}`)
+    }
+
+    if (options.maxCompressedResponseSize !== null && options.maxCompressedResponseSize > buffer.constants.MAX_LENGTH) {
+      throw new ConfigurationError(`The maxCompressedResponseSize cannot be bigger than ${buffer.constants.MAX_LENGTH}`)
+    }
+
     if (options.caFingerprint !== null && isHttpConnection(opts.node || opts.nodes)) {
       throw new ConfigurationError('You can\'t configure the caFingerprint with a http connection')
     }
@@ -178,7 +189,9 @@ class Client extends ESAPI {
       generateRequestId: options.generateRequestId,
       name: options.name,
       opaqueIdPrefix: options.opaqueIdPrefix,
-      context: options.context
+      context: options.context,
+      maxResponseSize: options.maxResponseSize,
+      maxCompressedResponseSize: options.maxCompressedResponseSize
     })
 
     this.helpers = new Helpers({

lib/Transport.d.ts

@@ -61,6 +61,8 @@ interface TransportOptions {
   generateRequestId?: generateRequestIdFn;
   name?: string;
   opaqueIdPrefix?: string;
+  maxResponseSize?: number;
+  maxCompressedResponseSize?: number;
 }
 
 export interface RequestEvent<TResponse = Record<string, any>, TContext = Context> {
@@ -113,6 +115,8 @@ export interface TransportRequestOptions {
   context?: Context;
   warnings?: string[];
   opaqueId?: string;
+  maxResponseSize?: number;
+  maxCompressedResponseSize?: number;
 }
 
 export interface TransportRequestCallback {

lib/Transport.js

@@ -43,6 +43,8 @@ const MAX_STRING_LENGTH = buffer.constants.MAX_STRING_LENGTH
 const kProductCheck = Symbol('product check')
 const kApiVersioning = Symbol('api versioning')
 const kEventEmitter = Symbol('event emitter')
+const kMaxResponseSize = Symbol('max response size')
+const kMaxCompressedResponseSize = Symbol('max compressed response size')
 
 class Transport {
   constructor (opts) {
@@ -72,6 +74,8 @@ class Transport {
     this[kProductCheck] = 0 // 0 = to be checked, 1 = checking, 2 = checked-ok, 3 checked-notok, 4 checked-nodefault
     this[kApiVersioning] = process.env.ELASTIC_CLIENT_APIVERSIONING === 'true'
     this[kEventEmitter] = new EventEmitter()
+    this[kMaxResponseSize] = opts.maxResponseSize || MAX_STRING_LENGTH
+    this[kMaxCompressedResponseSize] = opts.maxCompressedResponseSize || MAX_BUFFER_LENGTH
 
     this.nodeFilter = opts.nodeFilter || defaultNodeFilter
     if (typeof opts.nodeSelector === 'function') {
@@ -162,6 +166,8 @@ class Transport {
       ? 0
       : (typeof options.maxRetries === 'number' ? options.maxRetries : this.maxRetries)
     const compression = options.compression !== undefined ? options.compression : this.compression
+    const maxResponseSize = options.maxResponseSize || this[kMaxResponseSize]
+    const maxCompressedResponseSize = options.maxCompressedResponseSize || this[kMaxCompressedResponseSize]
     let request = { abort: noop }
     const transportReturn = {
       then (onFulfilled, onRejected) {
@@ -244,15 +250,15 @@ class Transport {
       /* istanbul ignore else */
       if (result.headers['content-length'] !== undefined) {
         const contentLength = Number(result.headers['content-length'])
-        if (isCompressed && contentLength > MAX_BUFFER_LENGTH) {
+        if (isCompressed && contentLength > maxCompressedResponseSize) {
           response.destroy()
           return onConnectionError(
-            new RequestAbortedError(`The content length (${contentLength}) is bigger than the maximum allowed buffer (${MAX_BUFFER_LENGTH})`, result)
+            new RequestAbortedError(`The content length (${contentLength}) is bigger than the maximum allowed buffer (${maxCompressedResponseSize})`, result)
           )
-        } else if (contentLength > MAX_STRING_LENGTH) {
+        } else if (contentLength > maxResponseSize) {
           response.destroy()
           return onConnectionError(
-            new RequestAbortedError(`The content length (${contentLength}) is bigger than the maximum allowed string (${MAX_STRING_LENGTH})`, result)
+            new RequestAbortedError(`The content length (${contentLength}) is bigger than the maximum allowed string (${maxResponseSize})`, result)
           )
         }
       }

test/unit/client.test.js

@@ -1308,6 +1308,223 @@ test('Content length too big (string)', t => {
   })
 })
 
+test('Content length too big custom (buffer)', t => {
+  t.plan(4)
+
+  class MockConnection extends Connection {
+    request (params, callback) {
+      const stream = intoStream(JSON.stringify({ hello: 'world' }))
+      stream.statusCode = 200
+      stream.headers = {
+        'content-type': 'application/json;utf=8',
+        'content-encoding': 'gzip',
+        'content-length': 1100,
+        connection: 'keep-alive',
+        date: new Date().toISOString()
+      }
+      stream.on('close', () => t.pass('Stream destroyed'))
+      process.nextTick(callback, null, stream)
+      return { abort () {} }
+    }
+  }
+
+  const client = new Client({
+    node: 'http://localhost:9200',
+    Connection: MockConnection,
+    maxCompressedResponseSize: 1000
+  })
+  client.info((err, result) => {
+    t.ok(err instanceof errors.RequestAbortedError)
+    t.equal(err.message, 'The content length (1100) is bigger than the maximum allowed buffer (1000)')
+    t.equal(result.meta.attempts, 0)
+  })
+})
+
+test('Content length too big custom (string)', t => {
+  t.plan(4)
+
+  class MockConnection extends Connection {
+    request (params, callback) {
+      const stream = intoStream(JSON.stringify({ hello: 'world' }))
+      stream.statusCode = 200
+      stream.headers = {
+        'content-type': 'application/json;utf=8',
+        'content-length': 1100,
+        connection: 'keep-alive',
+        date: new Date().toISOString()
+      }
+      stream.on('close', () => t.pass('Stream destroyed'))
+      process.nextTick(callback, null, stream)
+      return { abort () {} }
+    }
+  }
+
+  const client = new Client({
+    node: 'http://localhost:9200',
+    Connection: MockConnection,
+    maxResponseSize: 1000
+  })
+  client.info((err, result) => {
+    t.ok(err instanceof errors.RequestAbortedError)
+    t.equal(err.message, 'The content length (1100) is bigger than the maximum allowed string (1000)')
+    t.equal(result.meta.attempts, 0)
+  })
+})
+
+test('Content length too big custom option (buffer)', t => {
+  t.plan(4)
+
+  class MockConnection extends Connection {
+    request (params, callback) {
+      const stream = intoStream(JSON.stringify({ hello: 'world' }))
+      stream.statusCode = 200
+      stream.headers = {
+        'content-type': 'application/json;utf=8',
+        'content-encoding': 'gzip',
+        'content-length': 1100,
+        connection: 'keep-alive',
+        date: new Date().toISOString()
+      }
+      stream.on('close', () => t.pass('Stream destroyed'))
+      process.nextTick(callback, null, stream)
+      return { abort () {} }
+    }
+  }
+
+  const client = new Client({
+    node: 'http://localhost:9200',
+    Connection: MockConnection
+  })
+  client.info({}, { maxCompressedResponseSize: 1000 }, (err, result) => {
+    t.ok(err instanceof errors.RequestAbortedError)
+    t.equal(err.message, 'The content length (1100) is bigger than the maximum allowed buffer (1000)')
+    t.equal(result.meta.attempts, 0)
+  })
+})
+
+test('Content length too big custom option (string)', t => {
+  t.plan(4)
+
+  class MockConnection extends Connection {
+    request (params, callback) {
+      const stream = intoStream(JSON.stringify({ hello: 'world' }))
+      stream.statusCode = 200
+      stream.headers = {
+        'content-type': 'application/json;utf=8',
+        'content-length': 1100,
+        connection: 'keep-alive',
+        date: new Date().toISOString()
+      }
+      stream.on('close', () => t.pass('Stream destroyed'))
+      process.nextTick(callback, null, stream)
+      return { abort () {} }
+    }
+  }
+
+  const client = new Client({
+    node: 'http://localhost:9200',
+    Connection: MockConnection
+  })
+  client.info({}, { maxResponseSize: 1000 }, (err, result) => {
+    t.ok(err instanceof errors.RequestAbortedError)
+    t.equal(err.message, 'The content length (1100) is bigger than the maximum allowed string (1000)')
+    t.equal(result.meta.attempts, 0)
+  })
+})
+
+test('Content length too big custom option override (buffer)', t => {
+  t.plan(4)
+
+  class MockConnection extends Connection {
+    request (params, callback) {
+      const stream = intoStream(JSON.stringify({ hello: 'world' }))
+      stream.statusCode = 200
+      stream.headers = {
+        'content-type': 'application/json;utf=8',
+        'content-encoding': 'gzip',
+        'content-length': 1100,
+        connection: 'keep-alive',
+        date: new Date().toISOString()
+      }
+      stream.on('close', () => t.pass('Stream destroyed'))
+      process.nextTick(callback, null, stream)
+      return { abort () {} }
+    }
+  }
+
+  const client = new Client({
+    node: 'http://localhost:9200',
+    Connection: MockConnection,
+    maxCompressedResponseSize: 2000
+  })
+  client.info({}, { maxCompressedResponseSize: 1000 }, (err, result) => {
+    t.ok(err instanceof errors.RequestAbortedError)
+    t.equal(err.message, 'The content length (1100) is bigger than the maximum allowed buffer (1000)')
+    t.equal(result.meta.attempts, 0)
+  })
+})
+
+test('Content length too big custom option override (string)', t => {
+  t.plan(4)
+
+  class MockConnection extends Connection {
+    request (params, callback) {
+      const stream = intoStream(JSON.stringify({ hello: 'world' }))
+      stream.statusCode = 200
+      stream.headers = {
+        'content-type': 'application/json;utf=8',
+        'content-length': 1100,
+        connection: 'keep-alive',
+        date: new Date().toISOString()
+      }
+      stream.on('close', () => t.pass('Stream destroyed'))
+      process.nextTick(callback, null, stream)
+      return { abort () {} }
+    }
+  }
+
+  const client = new Client({
+    node: 'http://localhost:9200',
+    Connection: MockConnection,
+    maxResponseSize: 2000
+  })
+  client.info({}, { maxResponseSize: 1000 }, (err, result) => {
+    t.ok(err instanceof errors.RequestAbortedError)
+    t.equal(err.message, 'The content length (1100) is bigger than the maximum allowed string (1000)')
+    t.equal(result.meta.attempts, 0)
+  })
+})
+
+test('maxResponseSize cannot be bigger than buffer.constants.MAX_STRING_LENGTH', t => {
+  t.plan(2)
+
+  try {
+    new Client({ // eslint-disable-line
+      node: 'http://localhost:9200',
+      maxResponseSize: buffer.constants.MAX_STRING_LENGTH + 10
+    })
+    t.fail('should throw')
+  } catch (err) {
+    t.ok(err instanceof errors.ConfigurationError)
+    t.equal(err.message, `The maxResponseSize cannot be bigger than ${buffer.constants.MAX_STRING_LENGTH}`)
+  }
+})
+
+test('maxCompressedResponseSize cannot be bigger than buffer.constants.MAX_STRING_LENGTH', t => {
+  t.plan(2)
+
+  try {
+    new Client({ // eslint-disable-line
+      node: 'http://localhost:9200',
+      maxCompressedResponseSize: buffer.constants.MAX_LENGTH + 10
+    })
+    t.fail('should throw')
+  } catch (err) {
+    t.ok(err instanceof errors.ConfigurationError)
+    t.equal(err.message, `The maxCompressedResponseSize cannot be bigger than ${buffer.constants.MAX_LENGTH}`)
+  }
+})
+
 test('Meta header enabled', t => {
   t.plan(2)