zlib: reject reset while a zstd frame is incomplete

Resetting a ZstdCompress stream while a frame is still in
progress dropped the frame state, but any bytes already
written out stayed at the start of the output stream. The
next frame was then appended to that fragment, so the
resulting stream could not be decompressed. The failure was
silent: the compressor reported no error at all.

ZSTD_reset_session_only cancels unflushed internal data, so
refuse reset only when the incomplete frame has already
emitted output.

Signed-off-by: Xia Chao <shapirolutts@gmail.com>
PR-URL: #66088
Reviewed-By: James M Snell <jasnell@gmail.com>

Author: xia-chao

doc/api/zlib.md

@@ -2196,6 +2196,12 @@ configured for a Zstd compressor, it applies again to the next frame.
 
 Calling `reset()` while a write is in progress throws an `Error`.
 
+Resetting an incomplete Zstd compression frame after it has emitted output
+causes the stream to error with `ERR_ZLIB_INCOMPLETE_FRAME`. Resetting at
+that point would discard the frame state while the bytes already written
+out remain at the start of the output stream, leaving it undecodable. Call
+`.end()`, or start over with a new stream, instead.
+
 ## Class: `ZstdOptions`
 
 > Stability: 1 - Experimental

src/node_zlib.cc

@@ -352,6 +352,14 @@ class ZstdCompressContext final : public ZstdContext {
 
   uint64_t pledged_src_size_ = ZSTD_CONTENTSIZE_UNKNOWN;
   std::optional<uint64_t> consumed_src_size_;
+
+  // A frame is complete once ZSTD_compressStream2() has been called with
+  // ZSTD_e_end and has returned 0. Resetting an incomplete frame is only unsafe
+  // once some of that frame has already been written out: those bytes cannot be
+  // discarded, and the next frame would be appended to the fragment. Unflushed
+  // internal state alone is cancelled by ZSTD_reset_session_only.
+  bool frame_complete_ = true;
+  bool frame_output_emitted_ = false;
 };
 
 class ZstdDecompressContext final : public ZstdContext {
@@ -1678,6 +1686,8 @@ CompressionError ZstdCompressContext::Init(uint64_t pledged_src_size,
                                            std::string_view dictionary,
                                            bool) {
   pledged_src_size_ = pledged_src_size;
+  frame_complete_ = true;
+  frame_output_emitted_ = false;
   if (pledged_src_size == ZSTD_CONTENTSIZE_UNKNOWN) {
     consumed_src_size_.reset();
   } else {
@@ -1718,6 +1728,17 @@ CompressionError ZstdCompressContext::Init(uint64_t pledged_src_size,
 }
 
 CompressionError ZstdCompressContext::ResetStream() {
+  // ZSTD_reset_session_only cancels unflushed internal data. Bytes that have
+  // already been written out cannot be taken back, so refuse reset only when
+  // the current incomplete frame has already emitted output.
+  if (!frame_complete_ && frame_output_emitted_) {
+    return CompressionError(
+        "Cannot reset a zstd stream with an incomplete frame; end the frame "
+        "or discard the output produced so far",
+        "ERR_ZLIB_INCOMPLETE_FRAME",
+        ZSTD_error_stage_wrong);
+  }
+
   size_t result = ZSTD_CCtx_reset(cctx_.get(), ZSTD_reset_session_only);
   if (ZSTD_isError(result)) {
     const ZSTD_ErrorCode error = ZSTD_getErrorCode(result);
@@ -1737,6 +1758,8 @@ CompressionError ZstdCompressContext::ResetStream() {
   } else {
     consumed_src_size_ = 0;
   }
+  frame_complete_ = true;
+  frame_output_emitted_ = false;
   error_ = ZSTD_error_no_error;
   error_string_.clear();
   error_code_string_.clear();
@@ -1751,19 +1774,28 @@ void ZstdCompressContext::DoThreadPoolWork() {
   if (consumed_src_size_.has_value()) {
     *consumed_src_size_ += input_.pos - input_pos;
   }
+  if (output_.pos > 0) {
+    frame_output_emitted_ = true;
+  }
   if (ZSTD_isError(remaining)) {
     error_ = ZSTD_getErrorCode(remaining);
     error_code_string_ = ZstdStrerror(error_);
     error_string_ = ZSTD_getErrorString(error_);
-  } else if (remaining == 0 && flush_ == ZSTD_e_end &&
-             consumed_src_size_.has_value()) {
-    uint64_t const consumed_src_size = *consumed_src_size_;
-    consumed_src_size_.reset();
-    if (consumed_src_size != pledged_src_size_) {
-      error_ = ZSTD_error_srcSize_wrong;
-      error_code_string_ = ZstdStrerror(error_);
-      error_string_ = ZSTD_getErrorString(error_);
+    frame_complete_ = false;
+  } else if (remaining == 0 && flush_ == ZSTD_e_end) {
+    frame_complete_ = true;
+    frame_output_emitted_ = false;
+    if (consumed_src_size_.has_value()) {
+      uint64_t const consumed_src_size = *consumed_src_size_;
+      consumed_src_size_.reset();
+      if (consumed_src_size != pledged_src_size_) {
+        error_ = ZSTD_error_srcSize_wrong;
+        error_code_string_ = ZstdStrerror(error_);
+        error_string_ = ZSTD_getErrorString(error_);
+      }
     }
+  } else {
+    frame_complete_ = false;
   }
 }
 

test/parallel/test-zlib-zstd-reset-incomplete-frame.js

@@ -0,0 +1,129 @@
+'use strict';
+
+// Tests that reset() refuses to run only when an incomplete zstd frame has
+// already emitted output.
+//
+// ZSTD_reset_session_only cancels unflushed internal data, so write()-then-
+// reset() with no emitted bytes is safe. Once flush() (or a write that filled
+// the output buffer) has written a fragment out, resetting would append the
+// next frame to that fragment and produce an undecodable stream.
+
+require('../common');
+const assert = require('assert');
+const { finished } = require('stream/promises');
+const test = require('node:test');
+const zlib = require('zlib');
+
+test('ZstdCompress reset throws when an incomplete frame has emitted output',
+     async () => {
+       const stream = zlib.createZstdCompress();
+       const chunks = [];
+       stream.on('data', (chunk) => chunks.push(chunk));
+
+       stream.write(Buffer.from('hello'));
+       await new Promise((resolve) => stream.flush(resolve));
+       assert.ok(Buffer.concat(chunks).length > 0);
+
+       // A fragment of the frame is already outside the compressor, so reset()
+       // must refuse instead of silently producing a stream that cannot be
+       // decoded.
+       stream.reset();
+       stream.end(Buffer.from('world'));
+
+       await assert.rejects(finished(stream), {
+         code: 'ERR_ZLIB_INCOMPLETE_FRAME',
+       });
+     });
+
+test('ZstdCompress reset throws when write itself emitted frame output',
+     async () => {
+       const stream = zlib.createZstdCompress();
+       const chunks = [];
+       stream.on('data', (chunk) => chunks.push(chunk));
+
+       // Fill a buffer that is large enough for zstd to emit compressed bytes
+       // during write(), without an explicit flush().
+       const input = Buffer.allocUnsafe(512 * 1024);
+       for (let i = 0; i < input.length; i++) {
+         input[i] = i & 0xff;
+       }
+       await new Promise((resolve, reject) => {
+         stream.write(input, (err) => {
+           if (err) {
+             reject(err);
+           } else {
+             resolve();
+           }
+         });
+       });
+       assert.ok(Buffer.concat(chunks).length > 0);
+
+       stream.reset();
+       stream.end(Buffer.from('world'));
+
+       await assert.rejects(finished(stream), {
+         code: 'ERR_ZLIB_INCOMPLETE_FRAME',
+       });
+     });
+
+test('ZstdCompress reset after write without emitted output still works',
+     async () => {
+       const stream = zlib.createZstdCompress();
+       const chunks = [];
+       stream.on('data', (chunk) => chunks.push(chunk));
+
+       // Small writes often stay buffered inside zstd until flush/end.
+       await new Promise((resolve, reject) => {
+         stream.write(Buffer.from('hello'), (err) => {
+           if (err) {
+             reject(err);
+           } else {
+             resolve();
+           }
+         });
+       });
+       assert.strictEqual(Buffer.concat(chunks).length, 0);
+
+       // No bytes have left the compressor, so session reset is allowed.
+       stream.reset();
+       stream.end(Buffer.from('world'));
+       await finished(stream);
+
+       assert.strictEqual(
+         zlib.zstdDecompressSync(Buffer.concat(chunks)).toString(),
+         'world',
+       );
+     });
+
+test('ZstdCompress flush followed by end still produces a valid stream',
+     async () => {
+       const stream = zlib.createZstdCompress();
+       const chunks = [];
+       stream.on('data', (chunk) => chunks.push(chunk));
+
+       stream.write(Buffer.from('hello'));
+       await new Promise((resolve) => stream.flush(resolve));
+       stream.end(Buffer.from('world'));
+       await finished(stream);
+
+       assert.strictEqual(
+         zlib.zstdDecompressSync(Buffer.concat(chunks)).toString(),
+         'helloworld',
+       );
+     });
+
+test('ZstdCompress reset before any write still works', async () => {
+  const stream = zlib.createZstdCompress();
+  const chunks = [];
+  stream.on('data', (chunk) => chunks.push(chunk));
+
+  // No frame has been started yet, so reset() is allowed.
+  stream.reset();
+  stream.end(Buffer.from('hello'));
+  await finished(stream);
+
+  assert.strictEqual(
+    zlib.zstdDecompressSync(Buffer.concat(chunks)).toString(),
+    'hello',
+  );
+});