diff --git a/doc/api/zlib.md b/doc/api/zlib.md index 46b8eef1ad9f45..da863820957795 100644 --- a/doc/api/zlib.md +++ b/doc/api/zlib.md @@ -1100,6 +1100,9 @@ added: - v23.8.0 - v22.15.0 changes: + - version: REPLACEME + pr-url: https://github.com/nodejs/node/pull/64748 + description: Multiple concatenated zstd frames are decoded now. - version: REPLACEME pr-url: https://github.com/nodejs/node/pull/64599 description: The `dictionary` option can be a `TypedArray`, `DataView`, or @@ -1124,7 +1127,10 @@ Each Zstd-based class takes an `options` object. All options are optional. to improve compression efficiency when compressing or decompressing data that shares common patterns with the dictionary. * `rejectGarbageAfterEnd` {boolean} If `true`, decompression fails when - input remains after the first complete compressed stream. **Default:** `false` + trailing input is detected after the end of the compressed stream. This + includes unreadable bytes and additional zstd frames following the first + frame, which are otherwise decoded as part of the same stream. + **Default:** `false` For example: diff --git a/lib/zlib.js b/lib/zlib.js index e6c5c420d55182..0757e0a6b1cffd 100644 --- a/lib/zlib.js +++ b/lib/zlib.js @@ -949,6 +949,7 @@ class Zstd extends ZlibBase { writeState, processCallback, dictionary, + opts?.rejectGarbageAfterEnd === true, ); super(opts, mode, handle, zstdDefaultOpts); diff --git a/src/node_zlib.cc b/src/node_zlib.cc index 7d2950aed45a2d..8181802266da95 100644 --- a/src/node_zlib.cc +++ b/src/node_zlib.cc @@ -307,6 +307,7 @@ class ZstdContext : public MemoryRetainer { // Streaming-related, should be available for all compression libraries: void SetBuffers(const char* in, uint32_t in_len, char* out, uint32_t out_len); void SetFlush(int flush); + void SetRejectGarbageAfterEnd(bool reject_garbage_after_end); void GetAfterWriteOffsets(uint32_t* avail_in, uint32_t* avail_out) const; CompressionError GetErrorInfo() const; @@ -315,6 +316,7 @@ class ZstdContext : public MemoryRetainer { protected: ZSTD_EndDirective flush_ = ZSTD_e_continue; + bool reject_garbage_after_end_ = false; ZSTD_inBuffer input_ = {nullptr, 0, 0}; ZSTD_outBuffer output_ = {nullptr, 0, 0}; @@ -941,9 +943,9 @@ class ZstdStream final : public CompressionStream { } static void Init(const FunctionCallbackInfo& args) { - CHECK((args.Length() == 4 || args.Length() == 5) && - "init(params, pledgedSrcSize, writeResult, writeCallback[, " - "dictionary])"); + CHECK((args.Length() == 5 || args.Length() == 6) && + "init(params, pledgedSrcSize, writeResult, writeCallback, " + "dictionary[, rejectGarbageAfterEnd])"); ZstdStream* wrap; ASSIGN_OR_RETURN_UNWRAP(&wrap, args.This()); @@ -980,7 +982,7 @@ class ZstdStream final : public CompressionStream { AllocScope alloc_scope(wrap); std::string_view dictionary; ArrayBufferViewContents contents; - if (args.Length() == 5 && !args[4]->IsUndefined()) { + if (!args[4]->IsUndefined()) { if (!args[4]->IsArrayBufferView()) { THROW_ERR_INVALID_ARG_TYPE( wrap->env(), "dictionary must be an ArrayBufferView if provided"); @@ -997,6 +999,11 @@ class ZstdStream final : public CompressionStream { return; } + if (args.Length() == 6) { + CHECK(args[5]->IsBoolean()); + wrap->context()->SetRejectGarbageAfterEnd(args[5]->IsTrue()); + } + CHECK(args[0]->IsUint32Array()); const uint32_t* data = reinterpret_cast(Buffer::Data(args[0])); size_t len = args[0].As()->Length(); @@ -1624,6 +1631,10 @@ void ZstdContext::SetFlush(int flush) { flush_ = static_cast(flush); } +void ZstdContext::SetRejectGarbageAfterEnd(bool reject_garbage_after_end) { + reject_garbage_after_end_ = reject_garbage_after_end; +} + void ZstdContext::GetAfterWriteOffsets(uint32_t* avail_in, uint32_t* avail_out) const { *avail_in = input_.size - input_.pos; @@ -1765,15 +1776,21 @@ void ZstdDecompressContext::DoThreadPoolWork() { return; } - size_t const ret = ZSTD_decompressStream(dctx_.get(), &output_, &input_); - if (ZSTD_isError(ret)) { - frame_complete_ = false; - error_ = ZSTD_getErrorCode(ret); - error_code_string_ = ZstdStrerror(error_); - error_string_ = ZSTD_getErrorString(error_); - } else { + do { + size_t const ret = ZSTD_decompressStream(dctx_.get(), &output_, &input_); + if (ZSTD_isError(ret)) { + frame_complete_ = false; + error_ = ZSTD_getErrorCode(ret); + error_code_string_ = ZstdStrerror(error_); + error_string_ = ZSTD_getErrorString(error_); + return; + } frame_complete_ = ret == 0; - } + // A single input buffer may hold several concatenated frames. Keep decoding + // while one frame ends with input still pending and output space to fill, + // unless trailing input is meant to be rejected as garbage. + } while (frame_complete_ && !reject_garbage_after_end_ && + input_.pos < input_.size && output_.pos < output_.size); } CompressionError ZstdDecompressContext::GetErrorInfo() const { diff --git a/test/parallel/test-zlib-from-concatenated-zstd.js b/test/parallel/test-zlib-from-concatenated-zstd.js new file mode 100644 index 00000000000000..5c6b25b011f0bf --- /dev/null +++ b/test/parallel/test-zlib-from-concatenated-zstd.js @@ -0,0 +1,47 @@ +'use strict'; +// Test decompressing a zstd stream that contains multiple concatenated frames + +const common = require('../common'); +const assert = require('assert'); +const zlib = require('zlib'); + +const abc = 'abc'; +const def = 'def'; + +const abcEncoded = zlib.zstdCompressSync(abc); +const defEncoded = zlib.zstdCompressSync(def); + +const data = Buffer.concat([ + abcEncoded, + defEncoded, +]); + +assert.strictEqual(zlib.zstdDecompressSync(data).toString(), (abc + def)); + +zlib.zstdDecompress(data, common.mustSucceed((result) => { + assert.strictEqual(result.toString(), (abc + def)); +})); + +// Test that the next zstd frame can wrap around the input buffer boundary +[0, 1, 2, 3, 4, defEncoded.length].forEach((offset) => { + const resultBuffers = []; + + const decompress = zlib.createZstdDecompress() + .on('error', common.mustNotCall()) + .on('data', (data) => resultBuffers.push(data)) + .on('finish', common.mustCall(() => { + assert.strictEqual( + Buffer.concat(resultBuffers).toString(), + 'abcdef', + `result should match original input (offset = ${offset})` + ); + })); + + // First write: write "abc" + the first bytes of "def" + decompress.write(Buffer.concat([ + abcEncoded, defEncoded.slice(0, offset), + ])); + + // Write remaining bytes of "def" + decompress.end(defEncoded.slice(offset)); +}); diff --git a/test/parallel/test-zlib-reject-garbage-after-end.js b/test/parallel/test-zlib-reject-garbage-after-end.js index 8039865f5f1193..8d1abc118b37b4 100644 --- a/test/parallel/test-zlib-reject-garbage-after-end.js +++ b/test/parallel/test-zlib-reject-garbage-after-end.js @@ -78,7 +78,7 @@ const cases = [ decompress: zlib.zstdDecompress, decompressSync: zlib.zstdDecompressSync, createDecompress: zlib.createZstdDecompress, - defaultOutput: 'a', + defaultOutput: 'aa', }, ];