iemejia commented on code in PR #3852:
URL: https://github.com/apache/avro/pull/3852#discussion_r3568100201
##########
lang/js/test/test_files.js:
##########
@@ -531,6 +531,110 @@ 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('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);
+ });
+
Review Comment:
Added — a new 'normalizes decompress limits' test exercises the
finite-but-over-cap path (`normalize(cap + 1) === cap`,
`normalize(Number.MAX_VALUE) === cap`) plus flooring, string parsing, and the
invalid/out-of-range fallbacks.
normalizeMaxDecompressLength/MAX_DECOMPRESS_LENGTH_CAP are now exported for
tests.
##########
lang/js/test/test_files.js:
##########
@@ -531,6 +531,69 @@ 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 () { 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('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 () { cb(); });
Review Comment:
Already handled — the custom-codec test asserts the specific message
`/decompressed block size exceeds/` rather than passing on any error
(test_files.js:611).
##########
lang/js/test/test_files.js:
##########
@@ -531,6 +531,98 @@ 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('clamps an out-of-range limit instead of throwing', function (cb) {
Review Comment:
Already handled — the test is now named 'falls back to the default limit for
an out-of-range value' to match the fallback behavior.
##########
lang/js/test/test_files.js:
##########
@@ -531,6 +531,98 @@ 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.
+ 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('end', function () {
+ assert.deepEqual(out, [payload]);
+ cb();
+ });
+ encoder.pipe(decoder);
+ encoder.end(payload);
+ });
Review Comment:
Already handled — the test saves, clears, and restores
AVRO_MAX_DECOMPRESS_LENGTH so it's deterministic regardless of the runner
environment.
##########
lang/js/test/test_files.js:
##########
@@ -531,6 +531,109 @@ 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('end', function () {
+ restore();
+ assert.deepEqual(out, [payload]);
+ cb();
+ });
Review Comment:
Already handled — restore() runs on both the 'error' and 'end' paths
(test_files.js:586-588).
##########
lang/js/lib/files.js:
##########
@@ -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;
}
Review Comment:
Agreed this is worth doing, but calling destroy() interacts with the
hand-rolled finish/push(null) lifecycle in BlockDecoder and needs careful
testing, so it's out of scope for this decompression-limit PR. Tracked as a
follow-up: AVRO-4307 ("BlockDecoder should destroy itself on decompression
error to stop upstream I/O").
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
