node:zlib: reject invalid brotli flush values at validation

src/js/node/zlib.ts

@@ -195,6 +195,7 @@ function ZlibBase(opts, mode, handle, { flush, finishFlush, fullFlush }) {
   this._defaultFlushFlag = flush;
   this._finishFlushFlag = finishFlush;
   this._defaultFullFlushFlag = fullFlush;
+  this._flushBoundIdx = flushBoundIdx;
   this._info = opts && opts.info;
   this._maxOutputLength = maxOutputLength;
 }
@@ -268,6 +269,17 @@ ZlibBase.prototype.flush = function (kind, callback) {
     callback = kind;
     kind = this._defaultFullFlushFlag;
   }
+
+  // Reject kinds outside this codec's flush range before queuing the fake
+  // flush chunk — otherwise a brotli stream forwards a zlib-only flush value
+  // (Z_FINISH/Z_BLOCK) to the native encoder, which spins without progress
+  // (nodejs/node#63701). Matches Node's flush(kind) validation via
+  // checkRangesOrGetDefault: undefined/NaN fall through to the default,
+  // non-numbers throw ERR_INVALID_ARG_TYPE, out-of-range numbers throw
+  // ERR_OUT_OF_RANGE.
+  const flushBound = FLUSH_BOUND[this._flushBoundIdx];
+  kind = checkRangesOrGetDefault(kind, "kind", flushBound[0], flushBound[1], this._defaultFullFlushFlag);
+
   if (this.writableFinished) {
     if (callback) process.nextTick(callback);
   } else if (this.writableEnded) {
@@ -407,15 +419,25 @@ function processChunk(self, chunk, flushFlag, cb) {
   handle.inOff = 0;
   handle.flushFlag = flushFlag;
 
-  handle.write(
-    flushFlag, // flush
-    chunk, // in
-    0, // in_off
-    handle.availInBefore, // in_len
-    self._outBuffer, // out
-    self._outOffset, // out_off
-    handle.availOutBefore, // out_len
-  );
+  try {
+    handle.write(
+      flushFlag, // flush
+      chunk, // in
+      0, // in_off
+      handle.availInBefore, // in_len
+      self._outBuffer, // out
+      self._outOffset, // out_off
+      handle.availOutBefore, // out_len
+    );
+  } catch (err) {
+    // The native write rejects flush values the codec doesn't define (e.g.
+    // brotli with a zlib-only constant like Z_FINISH) synchronously, before
+    // scheduling any work. Route the error through the stream's error path
+    // instead of letting it escape into the stream machinery.
+    handle.buffer = null;
+    handle.cb = null;
+    cb(err);
+  }
 }
 
 function processCallback() {

src/runtime/node/node_zlib_binding.rs

@@ -182,9 +182,20 @@
 /// Backing-stream surface used by [`CompressionStream`] (zlib / brotli / zstd
 /// `Context` types).
 pub(crate) trait CompressionContext {
+    /// The codec's flush-operation type (brotli `BrotliEncoderOperation`, zlib
+    /// `FlushValue`, …). Holding a value of this type is proof the flush mode
+    /// is valid for this codec — invalid ints can't be converted, so they
+    /// can't reach the encoder.
+    type FlushOp: Copy;
+
     fn set_buffers(&mut self, in_: Option<&[u8]>, out: Option<&mut [u8]>);
-    fn set_flush(&mut self, flush: i32);
-    fn flush_value_is_valid(flush: u32) -> bool;
+    /// Convert a raw flush int (from JS) into this codec's flush op, or `None`
+    /// if it isn't a valid op for this codec. This is the single validation
+    /// point: brotli only has `BrotliEncoderOperation` 0..=3, so a zlib-only
+    /// mode like Z_FINISH(4)/Z_BLOCK(5) fails here instead of reaching the
+    /// encoder (where it would spin — see nodejs/node#63701).
+    fn flush_op_from_u32(flush: u32) -> Option<Self::FlushOp>;
+    fn set_flush(&mut self, op: Self::FlushOp);
     fn do_work(&mut self);
     fn reset(&mut self) -> Error;
     fn close(&mut self);
@@ -332,14 +343,19 @@
                 .throw());
         }
         let flush: u32 = jsv_to_u32(arguments[0]);
-        if !<T::Stream as CompressionContext>::flush_value_is_valid(flush) {
+        // Convert to the codec's typed flush op up front — failure means the
+        // value isn't a flush mode this codec defines. zlib.ts's `.flush()`
+        // already rejects out-of-range kinds (ERR_OUT_OF_RANGE) before queuing
+        // the flush chunk; this is defense-in-depth for direct handle.write()
+        // and `_processChunk()` callers that bypass `.flush()`.
+        let Some(flush_op) = <T::Stream as CompressionContext>::flush_op_from_u32(flush) else {
             return Err(global_this
                 .err(
                     ErrorCode::INVALID_ARG_VALUE,
                     format_args!("Invalid flush value"),
                 )
                 .throw());
-        }
+        };
 
         if arguments[1].is_null() {
             // just a flush
@@ -451,7 +467,7 @@
 
         this.stream().with_mut(|s| {
             s.set_buffers(in_, out);
-            s.set_flush(i32::try_from(flush).expect("int cast"));
+            s.set_flush(flush_op);
         });
 
         // Only create the strong handle when we have a pending write
@@ -648,14 +664,19 @@
                 .throw());
         }
         let flush: u32 = jsv_to_u32(arguments[0]);
-        if !<T::Stream as CompressionContext>::flush_value_is_valid(flush) {
+        // Convert to the codec's typed flush op up front — failure means the
+        // value isn't a flush mode this codec defines. zlib.ts's `.flush()`
+        // already rejects out-of-range kinds (ERR_OUT_OF_RANGE) before queuing
+        // the flush chunk; this is defense-in-depth for direct handle.write()
+        // and `_processChunk()` callers that bypass `.flush()`.
+        let Some(flush_op) = <T::Stream as CompressionContext>::flush_op_from_u32(flush) else {
             return Err(global_this
                 .err(
                     ErrorCode::INVALID_ARG_VALUE,
                     format_args!("Invalid flush value"),
                 )
                 .throw());
-        }
+        };
 
         // Hoisted so `in_` can borrow it past the `else` arm (mirrors `out_buf`).
         let in_buf: jsc::ArrayBuffer;
@@ -730,7 +751,7 @@
 
         this.stream().with_mut(|s| {
             s.set_buffers(in_, out);
-            s.set_flush(i32::try_from(flush).expect("int cast"));
+            s.set_flush(flush_op);
         });
         let this_value = callframe.this();
 
@@ -1014,10 +1035,14 @@
 /// emits a `pub mod js { … }` with the cached-property accessors
 /// (`writeCallback` / `errorCallback` / `dictionary`) wired to the
 /// `${TypeName}Prototype__${prop}{Get,Set}CachedValue` extern symbols.
+///
+/// `$flush_op` is the codec's [`CompressionContext::FlushOp`] — the typed
+/// flush operation its fallible `flush_op_from_u32` produces and its
+/// `set_flush` consumes.
 #[macro_export]
 #[doc(hidden)]
 macro_rules! __impl_compression_stream {
-    ($native:ident, $ctx:ty, $type_name:literal) => {
+    ($native:ident, $ctx:ty, $type_name:literal, $flush_op:ty) => {
         impl ::bun_event_loop::Taskable for $native {
             const TAG: ::bun_event_loop::TaskTag = ::bun_event_loop::task_tag::$native;
             /// An async write whose completion will not run: unpin, unref, drop
@@ -1036,9 +1061,11 @@
         }
 
         impl $crate::node::node_zlib_binding::CompressionContext for $ctx {
+            type FlushOp = $flush_op;
+
             #[inline] fn set_buffers(&mut self, in_: Option<&[u8]>, out: Option<&mut [u8]>) { Self::set_buffers(self, in_, out) }
-            #[inline] fn set_flush(&mut self, flush: i32) { Self::set_flush(self, flush) }
-            #[inline] fn flush_value_is_valid(flush: u32) -> bool { Self::flush_value_is_valid(flush) }
+            #[inline] fn flush_op_from_u32(flush: u32) -> Option<$flush_op> { Self::flush_op_from_u32(flush) }
+            #[inline] fn set_flush(&mut self, op: $flush_op) { Self::set_flush(self, op) }
             #[inline] fn do_work(&mut self) { Self::do_work(self) }
             #[inline] fn reset(&mut self) -> $crate::node::node_zlib_binding::Error { Self::reset(self) }
             #[inline] fn close(&mut self) { Self::close(self) }

src/runtime/node/zlib/NativeBrotli.rs

@@ -541,22 +541,22 @@ mod _impl {
             }
         }
 
-        pub fn flush_value_is_valid(flush: u32) -> bool {
-            flush <= 3
+        /// The four real `BrotliEncoderOperation`s are the only valid brotli
+        /// flush values. A zlib-only mode (Z_FINISH=4/Z_BLOCK=5/Z_TREES=6) has
+        /// no brotli op, so it fails the conversion and the shared write path
+        /// rejects it before it can reach the encoder.
+        pub fn flush_op_from_u32(flush: u32) -> Option<Op> {
+            match flush {
+                0 => Some(Op::process),
+                1 => Some(Op::flush),
+                2 => Some(Op::finish),
+                3 => Some(Op::emit_metadata),
+                _ => None,
+            }
         }
 
-        pub fn set_flush(&mut self, flush: c_int) {
-            // Caller passes a valid BrotliEncoderOperation discriminant (Node
-            // zlib constants 0..=3). Exhaustive match — `Op` is `#[repr(u32)]`
-            // so the prior `c_int` bit-cast was a width hazard anyway. Out-of-
-            // range traps.
-            self.flush = match flush {
-                0 => Op::process,
-                1 => Op::flush,
-                2 => Op::finish,
-                3 => Op::emit_metadata,
-                n => unreachable!("invalid BrotliEncoderOperation {n}"),
-            };
+        pub fn set_flush(&mut self, op: Op) {
+            self.flush = op;
         }
 
         pub fn do_work(&mut self) {
@@ -700,7 +700,11 @@ mod _impl {
     // Stamps `impl CompressionContext for Context`, `impl Taskable`/
     // `CompressionStreamImpl for NativeBrotli`, and `pub mod js { … }` (the
     // `NativeBrotliPrototype__*CachedValue` accessors).
-    crate::__impl_compression_stream!(NativeBrotli, Context, "NativeBrotli");
+    //
+    // FlushOp = `BrotliEncoderOperation` (PROCESS/FLUSH/FINISH/EMIT_METADATA).
+    // Zlib-only flush values (Z_FINISH=4/Z_BLOCK=5/Z_TREES=6) have no brotli op
+    // and are rejected at the write boundary by `flush_op_from_u32`.
+    crate::__impl_compression_stream!(NativeBrotli, Context, "NativeBrotli", Op);
 
     fn code_for_error(err: c::BrotliDecoderErrorCode2) -> *const core::ffi::c_char {
         // Node builds these as `"ERR_" + <brotli enum suffix>` where the suffix

src/runtime/node/zlib/NativeZlib.rs

@@ -272,7 +272,8 @@ mod _impl {
         }
     }
 
-    crate::__impl_compression_stream!(NativeZlib, super::Context, "NativeZlib");
+    // FlushOp = the full zlib `FlushValue` range (Z_NO_FLUSH..=Z_TREES).
+    crate::__impl_compression_stream!(NativeZlib, super::Context, "NativeZlib", c::FlushValue);
     crate::__compression_stream_mixin_reexports!(NativeZlib);
 } // mod _impl
 
@@ -480,23 +481,25 @@ impl Context {
         };
     }
 
-    pub fn flush_value_is_valid(flush: u32) -> bool {
-        flush <= 6
+    /// zlib accepts the full `FlushValue` range (Z_NO_FLUSH..=Z_TREES).
+    /// Checked conversion — transmuting an arbitrary int into a Rust enum is
+    /// UB, and anything outside the range fails here so the shared write path
+    /// rejects it.
+    pub fn flush_op_from_u32(flush: u32) -> Option<c::FlushValue> {
+        match flush {
+            0 => Some(c::FlushValue::NoFlush),
+            1 => Some(c::FlushValue::PartialFlush),
+            2 => Some(c::FlushValue::SyncFlush),
+            3 => Some(c::FlushValue::FullFlush),
+            4 => Some(c::FlushValue::Finish),
+            5 => Some(c::FlushValue::Block),
+            6 => Some(c::FlushValue::Trees),
+            _ => None,
+        }
     }
 
-    pub fn set_flush(&mut self, flush: c_int) {
-        // Checked conversion;
-        // transmuting an arbitrary c_int into a Rust enum is UB.
-        self.flush = match flush {
-            0 => c::FlushValue::NoFlush,
-            1 => c::FlushValue::PartialFlush,
-            2 => c::FlushValue::SyncFlush,
-            3 => c::FlushValue::FullFlush,
-            4 => c::FlushValue::Finish,
-            5 => c::FlushValue::Block,
-            6 => c::FlushValue::Trees,
-            _ => unreachable!("invalid zlib flush value: {flush}"),
-        };
+    pub fn set_flush(&mut self, op: c::FlushValue) {
+        self.flush = op;
     }
 
     pub fn do_work(&mut self) {

src/runtime/node/zlib/NativeZstd.rs

@@ -497,12 +497,20 @@ mod _impl {
             self.output.pos = 0;
         }
 
-        pub fn flush_value_is_valid(flush: u32) -> bool {
-            flush <= 2
+        /// The flush value is forwarded to `ZSTD_compressStream2` as a
+        /// `ZSTD_EndDirective`, which only defines `ZSTD_e_continue` (0),
+        /// `ZSTD_e_flush` (1) and `ZSTD_e_end` (2); anything else fails here so
+        /// the shared write path rejects it.
+        pub fn flush_op_from_u32(flush: u32) -> Option<c_int> {
+            if flush <= 2 {
+                Some(flush as c_int)
+            } else {
+                None
+            }
         }
 
-        pub fn set_flush(&mut self, flush: c_int) {
-            self.flush = flush;
+        pub fn set_flush(&mut self, op: c_int) {
+            self.flush = op;
         }
 
         const ZSTD_MAGICNUMBER: [u8; 4] = 0xFD2FB528u32.to_le_bytes();
@@ -684,6 +692,8 @@ mod _impl {
     // `CompressionStreamImpl for NativeZstd`, and `pub mod js { … }` so
     // `CompressionStream::<NativeZstd>::*` (write/writeSync/reset/close/
     // emit_error/…) can reach this struct's fields.
-    crate::__impl_compression_stream!(NativeZstd, Context, "NativeZstd");
+    // FlushOp = raw `c_int` holding a `ZSTD_EndDirective` (0..=2), validated
+    // by `flush_op_from_u32`.
+    crate::__impl_compression_stream!(NativeZstd, Context, "NativeZstd", c_int);
     crate::__compression_stream_mixin_reexports!(NativeZstd);
 } // mod _impl