diff --git a/lang/js/lib/files.js b/lang/js/lib/files.js index 6d63cb45087..65f58d6ea1d 100644 --- a/lang/js/lib/files.js +++ b/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, diff --git a/lang/js/test/test_files.js b/lang/js/test/test_files.js index 43d5df60162..a39e2318ab3 100644 --- a/lang/js/test/test_files.js +++ b/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) {