diff --git a/doc/api/zlib.md b/doc/api/zlib.md index 9ab581678e08..e7a8a809d012 100644 --- a/doc/api/zlib.md +++ b/doc/api/zlib.md @@ -2202,6 +2202,9 @@ added: - v23.8.0 - v22.15.0 changes: + - version: REPLACEME + pr-url: https://github.com/nodejs/node/pull/65911 + description: Multiple concatenated Zstd frames in the input are now decoded. - version: - v26.7.0 - v24.20.0 @@ -2230,7 +2233,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 as well as additional Zstd frames following the + first one, 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 3e6986bf8127..24983be53457 100644 --- a/lib/zlib.js +++ b/lib/zlib.js @@ -924,6 +924,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 af82aa2ae73b..739fd3ddcca4 100644 --- a/src/node_zlib.cc +++ b/src/node_zlib.cc @@ -308,6 +308,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; @@ -316,6 +317,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}; @@ -947,9 +949,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()); @@ -986,7 +988,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"); @@ -996,6 +998,11 @@ class ZstdStream final : public CompressionStream { dictionary = std::string_view(contents.data(), contents.length()); } + if (args.Length() == 6) { + CHECK(args[5]->IsBoolean()); + wrap->context()->SetRejectGarbageAfterEnd(args[5]->IsTrue()); + } + CompressionError err = wrap->context()->Init(pledged_src_size, dictionary); if (err.IsError()) { wrap->EmitError(err); @@ -1630,6 +1637,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; @@ -1790,15 +1801,24 @@ 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 { + // A single input buffer may hold several concatenated frames, as the zstd + // format explicitly allows (see `zstdcat`). Keep decoding while a frame ends + // with input still pending and output space left to fill, mirroring how the + // gzip path handles concatenated members. When `rejectGarbageAfterEnd` is + // set, stop after the first frame so that the remaining input is reported as + // trailing junk. + 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; - } + } 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 000000000000..13e2f5108d24 --- /dev/null +++ b/test/parallel/test-zlib-from-concatenated-zstd.js @@ -0,0 +1,63 @@ +'use strict'; +// Test decompressing a zstd payload that contains multiple concatenated frames. +// The zstd format allows this and `zstdcat` decodes it, so the streaming and +// one-shot APIs should too. +// Refs: https://github.com/nodejs/node/issues/64741 + +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.subarray(0, offset), + ])); + + // Write remaining bytes of "def". + decompress.end(defEncoded.subarray(offset)); +}); + +// With `rejectGarbageAfterEnd`, a trailing frame is treated as junk: the first +// frame still decodes, but the stream then errors instead of decoding the rest. +{ + const chunks = []; + const decompress = zlib.createZstdDecompress({ rejectGarbageAfterEnd: true }) + .on('data', (chunk) => chunks.push(chunk)) + .on('error', common.mustCall((err) => { + assert.strictEqual(err.code, 'ERR_TRAILING_JUNK_AFTER_STREAM_END'); + assert.strictEqual(Buffer.concat(chunks).toString(), abc); + })); + decompress.end(data); +} diff --git a/test/parallel/test-zlib-reject-garbage-after-end.js b/test/parallel/test-zlib-reject-garbage-after-end.js index 8039865f5f11..8d1abc118b37 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', }, ];