zlib: decode concatenated zstd frames

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:
 

lib/zlib.js

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

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<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());
@@ -980,7 +982,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");
@@ -997,6 +999,11 @@ class ZstdStream final : public CompressionStream<CompressionContext> {
       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<uint32_t*>(Buffer::Data(args[0]));
     size_t len = args[0].As<Uint32Array>()->Length();
@@ -1624,6 +1631,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;
@@ -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 {

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

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