zlib: reject reset while a zstd frame is incomplete

doc/api/zlib.md

@@ -2200,6 +2200,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',
+  );
+});