AVRO-4287: [javascript] Enforce a maximum decompressed block size

lang/js/lib/files.js

@@ -24,6 +24,7 @@
 var protocols = require('./protocols'),
     schemas = require('./schemas'),
     utils = require('./utils'),
+    buffer = require('buffer'),
     fs = require('fs'),
     stream = require('stream'),
     util = require('util'),
@@ -58,6 +59,41 @@ var LONG_TYPE = schemas.createType('long');
 // First 4 bytes of an Avro object container file.
 var MAGIC_BYTES = Buffer.from('Obj\x01');
 
+// Default upper bound, in bytes, on the size a single data-file block may
+// decompress to. A block with a very high compression ratio (or a malformed
+// block) can otherwise expand to far more memory than its compressed size.
+// Mirrors the Java SDK's decompression limit (AVRO-4247). Overridable per
+// decoder via the `maxDecompressLength` option, or globally with the
+// `AVRO_MAX_DECOMPRESS_LENGTH` environment variable.
+var DEFAULT_MAX_DECOMPRESS_LENGTH = 200 * 1024 * 1024; // 200 MiB
+
+// Largest value safe to pass as zlib's `maxOutputLength`. A value above the
+// runtime's maximum buffer length (or a non-finite / non-integer one) can make
+// `zlib.inflateRaw` throw synchronously, so every configured limit is
+// normalized to a positive integer bounded by this cap before use.
+var MAX_DECOMPRESS_LENGTH_CAP =
+  (buffer.constants && buffer.constants.MAX_LENGTH) || 0x7fffffff;
+
+// Normalize a user- or environment-supplied limit to a positive, finite
+// integer bounded by MAX_DECOMPRESS_LENGTH_CAP. Returns undefined when the
+// input is missing or invalid, so callers can fall back to a default.
+function normalizeMaxDecompressLength(value) {
+  var parsed = typeof value === 'number' ? value : parseInt(value, 10);
+  if (!isFinite(parsed) || parsed <= 0) {
+    return undefined;
+  }
+  parsed = Math.floor(parsed);
+  if (parsed > MAX_DECOMPRESS_LENGTH_CAP) {
+    return MAX_DECOMPRESS_LENGTH_CAP;
+  }
+  return parsed;
+}
+
+function getDefaultMaxDecompressLength() {
+  var normalized = normalizeMaxDecompressLength(process.env.AVRO_MAX_DECOMPRESS_LENGTH);
+  return normalized === undefined ? DEFAULT_MAX_DECOMPRESS_LENGTH : normalized;
+}
+
 // Convenience.
 var f = util.format;
 var Tap = utils.Tap;
@@ -145,6 +181,9 @@ function BlockDecoder(opts) {
   this._type = null;
   this._codecs = opts.codecs;
   this._parseOpts = opts.parseOpts || {};
+  var normalizedMax = normalizeMaxDecompressLength(opts.maxDecompressLength);
+  this._maxDecompressLength = normalizedMax === undefined ?
+    getDefaultMaxDecompressLength() : normalizedMax;
   this._tap = new Tap(Buffer.alloc(0));
   this._blockTap = new Tap(Buffer.alloc(0));
   this._syncMarker = null;
@@ -187,12 +226,41 @@ BlockDecoder.prototype._decodeHeader = function () {
   }
 
   var codec = (header.meta['avro.codec'] || 'null').toString();
-  this._decompress = (this._codecs || BlockDecoder.getDefaultCodecs())[codec];
-  if (!this._decompress) {
+  var decompress = (this._codecs || BlockDecoder.getDefaultCodecs())[codec];
+  if (!decompress) {
     this.emit('error', new Error(f('unknown codec: %s', codec)));
     return;
   }
 
+  // Bound the decompressed size of each block to guard against a block with a
+  // very high compression ratio (or a malformed block) expanding to far more
+  // memory than its compressed size. Whenever the resolved codec is Node's raw
+  // inflater -- the built-in default or a custom codecs map that reuses it --
+  // the limit is passed to zlib so the allocation itself is capped; every
+  // codec's output is additionally checked as a safeguard.
+  var maxLength = this._maxDecompressLength;
+  if (decompress === zlib.inflateRaw) {
+    decompress = function (buf, cb) {
+      zlib.inflateRaw(buf, {maxOutputLength: maxLength}, cb);
+    };
+  }
+  this._decompress = function (buf, cb) {
+    decompress(buf, function (err, data) {
+      if (err) {
+        cb(err);
+        return;
+      }
+      if (data && data.length > maxLength) {
+        cb(new Error(f(
+          'decompressed block size exceeds the maximum allowed of %d bytes',
+          maxLength
+        )));
+        return;
+      }
+      cb(null, data);
+    });
+  };
+
   try {
     var schema = JSON.parse(header.meta['avro.schema'].toString());
     this._type = parse(schema, this._parseOpts);
@@ -247,11 +315,21 @@ BlockDecoder.prototype._createBlockCallback = function () {
   this._pending++;
 
   return function (err, data) {
+    // Always account for this block's completion, even on error, so a failed
+    // decompression (e.g. the size safeguard firing) cannot leave `_pending`
+    // permanently above zero and hang the stream at finish.
+    self._pending--;
     if (err) {
       self.emit('error', err);
+      // If the writable side already finished and this was the last pending
+      // block, end the readable side too so a consumer waiting on the queue is
+      // not left hanging after the error.
+      if (self._needPush && self._finished && !self._pending) {
+        self._needPush = false;
+        self.push(null);
+      }
       return;
     }
-    self._pending--;
     self._queue.push(new BlockData(index, data));
     if (self._needPush) {
       self._needPush = false;
@@ -663,6 +741,8 @@ function loadSchema(schema) {
 module.exports = {
   HEADER_TYPE: HEADER_TYPE, // For tests.
   MAGIC_BYTES: MAGIC_BYTES, // Idem.
+  MAX_DECOMPRESS_LENGTH_CAP: MAX_DECOMPRESS_LENGTH_CAP, // Idem.
+  normalizeMaxDecompressLength: normalizeMaxDecompressLength, // Idem.
   parse: parse,
   createFileDecoder: createFileDecoder,
   createFileEncoder: createFileEncoder,

lang/js/test/test_files.js

@@ -531,6 +531,128 @@ describe('files', function () {
         });
     });
 
+    it('rejects a block that decompresses beyond the limit', function (cb) {
+      // A large, highly compressible payload compresses to a tiny block but
+      // would decompress to far more than the configured limit.
+      var t = createType('string');
+      var big = new Array(100001).join('a'); // 100000 bytes when decompressed
+      var encoder = new streams.BlockEncoder(t, {codec: 'deflate'});
+      var decoder = new streams.BlockDecoder({maxDecompressLength: 1024})
+        .on('data', function () {})
+        .on('error', function (err) {
+          // zlib's maxOutputLength cap fires, so the allocation itself is bounded.
+          assert.equal(err.code, 'ERR_BUFFER_TOO_LARGE');
+          cb();
+        });
+      encoder.pipe(decoder);
+      encoder.end(big);
+    });
+
+    it('decompresses a block within the limit', function (cb) {
+      var t = createType('string');
+      var payload = 'hello world';
+      var out = [];
+      var encoder = new streams.BlockEncoder(t, {codec: 'deflate'});
+      var decoder = new streams.BlockDecoder({maxDecompressLength: 1024})
+        .on('data', function (s) { out.push(s); })
+        .on('end', function () {
+          assert.deepEqual(out, [payload]);
+          cb();
+        });
+      encoder.pipe(decoder);
+      encoder.end(payload);
+    });
+
+    it('falls back to the default limit for an out-of-range value', function (cb) {
+      // A non-finite/out-of-range limit is not usable, so normalizeMaxDecompressLength
+      // returns undefined and the decoder falls back to its default limit rather
+      // than making zlib throw synchronously; a normal block still decodes.
+      // Clear the env override so the fallback default is deterministic.
+      var savedEnv = process.env.AVRO_MAX_DECOMPRESS_LENGTH;
+      delete process.env.AVRO_MAX_DECOMPRESS_LENGTH;
+      var restore = function () {
+        if (savedEnv === undefined) {
+          delete process.env.AVRO_MAX_DECOMPRESS_LENGTH;
+        } else {
+          process.env.AVRO_MAX_DECOMPRESS_LENGTH = savedEnv;
+        }
+      };
+      var t = createType('string');
+      var payload = 'hello world';
+      var out = [];
+      var encoder = new streams.BlockEncoder(t, {codec: 'deflate'});
+      var decoder = new streams.BlockDecoder({maxDecompressLength: Infinity})
+        .on('data', function (s) { out.push(s); })
+        .on('error', function (err) { restore(); cb(err); })
+        .on('end', function () {
+          restore();
+          assert.deepEqual(out, [payload]);
+          cb();
+        });
+      encoder.pipe(decoder);
+      encoder.end(payload);
+    });
+
+    it('enforces the limit for custom codecs', function (cb) {
+      // A custom codec that yields more than the limit is rejected by the
+      // size safeguard applied to every codec.
+      var t = createType('int');
+      var codecs = {
+        'null': function (data, cb) { cb(null, Buffer.alloc(4096)); }
+      };
+      var encoder = new streams.BlockEncoder(t, {codec: 'null'});
+      var decoder = new streams.BlockDecoder({
+        codecs: codecs,
+        maxDecompressLength: 1024
+      })
+        .on('data', function () {})
+        .on('error', function (err) {
+          // The generic post-decompress size safeguard fires for custom codecs.
+          assert(/decompressed block size exceeds/.test(err.message));
+          cb();
+        });
+      encoder.pipe(decoder);
+      encoder.end(1);
+    });
+
+    it('normalizes decompress limits', function () {
+      var cap = files.MAX_DECOMPRESS_LENGTH_CAP;
+      var normalize = files.normalizeMaxDecompressLength;
+      // A finite value above the cap is clamped to the cap (the "> cap" path).
+      assert.equal(normalize(cap + 1), cap);
+      assert.equal(normalize(Number.MAX_VALUE), cap);
+      // A finite value at/below the cap is floored and kept.
+      assert.equal(normalize(1024), 1024);
+      assert.equal(normalize(1024.9), 1024);
+      assert.equal(normalize('2048'), 2048);
+      // Missing / invalid / out-of-range values fall back to undefined.
+      assert.equal(normalize(undefined), undefined);
+      assert.equal(normalize(Infinity), undefined);
+      assert.equal(normalize(0), undefined);
+      assert.equal(normalize(-1), undefined);
+      assert.equal(normalize('nope'), undefined);
+    });
+
+    it('caps a custom deflate codec that reuses zlib.inflateRaw', function (cb) {
+      // A custom codecs map that reuses the built-in raw inflater must still get
+      // the zlib maxOutputLength cap (not just the post-decompress check).
+      var zlib = require('zlib');
+      var t = createType('string');
+      var big = new Array(100001).join('a'); // 100000 bytes when decompressed
+      var encoder = new streams.BlockEncoder(t, {codec: 'deflate'});
+      var decoder = new streams.BlockDecoder({
+        codecs: {deflate: zlib.inflateRaw},
+        maxDecompressLength: 1024
+      })
+        .on('data', function () {})
+        .on('error', function (err) {
+          assert.equal(err.code, 'ERR_BUFFER_TOO_LARGE');
+          cb();
+        });
+      encoder.pipe(decoder);
+      encoder.end(big);
+    });
+
   });
 
   it('createFileDecoder', function (cb) {