zlib: decode concatenated zstd frames

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:
 

lib/zlib.js

@@ -924,6 +924,7 @@ class Zstd extends ZlibBase {
       writeState,
       processCallback,
       dictionary,
+      opts?.rejectGarbageAfterEnd === true,
     );
 
     super(opts, mode, handle, zstdDefaultOpts);

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<CompressionContext> {
   }
 
   static void Init(const FunctionCallbackInfo<Value>& 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<CompressionContext> {
     AllocScope alloc_scope(wrap);
     std::string_view dictionary;
     ArrayBufferViewContents<char> 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<CompressionContext> {
       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<ZSTD_EndDirective>(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 {

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);
+}

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',
   },
 ];