Repository: denoland/deno
Pull request: fix(ext/node): improve input validations of stream/consumers

This PR improves input validations of stream consumers APIs, and enables `parallel/test-stream-consumers.js` test case.

towards #29802 

Changed files:
filename: ext/node/polyfills/stream/consumers.js
patch: @@ -1,16 +1,32 @@
 // deno-lint-ignore-file
 // Copyright 2018-2025 the Deno authors. MIT license.
 
-import { primordials } from "ext:core/mod.js";
+import { core, primordials } from "ext:core/mod.js";
 import { TextDecoder } from "ext:deno_web/08_text_encoding.js";
 import { Blob } from "ext:deno_web/09_file.js";
 import { Buffer } from "node:buffer";
-"use strict";
+import {
+  ERR_INVALID_ARG_TYPE,
+  ERR_INVALID_STATE,
+} from "ext:deno_node/internal/errors.ts";
 
 const {
+  ArrayBufferIsView,
   JSONParse,
 } = primordials;
 
+function validateStreamIterator(stream) {
+  try {
+    return stream[Symbol.asyncIterator]();
+  } catch (e) {
+    if (e instanceof TypeError && e.message === "ReadableStream is locked") {
+      throw new ERR_INVALID_STATE(e.message);
+    } else {
+      throw e;
+    }
+  }
+}
+
 /**
  * @typedef {import('../internal/webstreams/readablestream').ReadableStream
  * } ReadableStream
@@ -23,7 +39,8 @@ const {
  */
 async function blob(stream) {
   const chunks = [];
-  for await (const chunk of stream) {
+  const iter = validateStreamIterator(stream);
+  for await (const chunk of iter) {
     chunks.push(chunk);
   }
   return new Blob(chunks);
@@ -53,10 +70,18 @@ async function buffer(stream) {
 async function text(stream) {
   const dec = new TextDecoder();
   let str = "";
-  for await (const chunk of stream) {
+  const iter = validateStreamIterator(stream);
+  for await (const chunk of iter) {
     if (typeof chunk === "string") {
       str += chunk;
     } else {
+      if (!core.isAnyArrayBuffer(chunk) && !ArrayBufferIsView(chunk)) {
+        throw new ERR_INVALID_ARG_TYPE("input", [
+          "SharedArrayBuffer",
+          "ArrayBuffer",
+          "ArrayBufferView",
+        ], chunk);
+      }
       str += dec.decode(chunk, { stream: true });
     }
   }

filename: tests/node_compat/config.jsonc
patch: @@ -1036,6 +1036,7 @@
       "test-stream-compose-operator.js",
       "test-stream-compose.js",
       "test-stream-construct.js",
+      "test-stream-consumers.js",
       "test-stream-decoder-objectmode.js",
       "test-stream-destroy-event-order.js",
       "test-stream-drop-take.js",

filename: tests/node_compat/runner/TODO.md
patch: @@ -2038,7 +2038,6 @@ NOTE: This file should not be manually edited. Please edit `tests/node_compat/co
 - [parallel/test-stdout-to-file.js](https://github.com/nodejs/node/tree/v23.9.0/test/parallel/test-stdout-to-file.js)
 - [parallel/test-strace-openat-openssl.js](https://github.com/nodejs/node/tree/v23.9.0/test/parallel/test-strace-openat-openssl.js)
 - [parallel/test-stream-base-typechecking.js](https://github.com/nodejs/node/tree/v23.9.0/test/parallel/test-stream-base-typechecking.js)
-- [parallel/test-stream-consumers.js](https://github.com/nodejs/node/tree/v23.9.0/test/parallel/test-stream-consumers.js)
 - [parallel/test-stream-destroy.js](https://github.com/nodejs/node/tree/v23.9.0/test/parallel/test-stream-destroy.js)
 - [parallel/test-stream-pipe-deadlock.js](https://github.com/nodejs/node/tree/v23.9.0/test/parallel/test-stream-pipe-deadlock.js)
 - [parallel/test-stream-pipeline-http2.js](https://github.com/nodejs/node/tree/v23.9.0/test/parallel/test-stream-pipeline-http2.js)

filename: tests/node_compat/test/parallel/test-stream-consumers.js
patch: @@ -0,0 +1,269 @@
+// deno-fmt-ignore-file
+// deno-lint-ignore-file
+
+// Copyright Joyent and Node contributors. All rights reserved. MIT license.
+// Taken from Node 23.9.0
+// This file is automatically generated by `tests/node_compat/runner/setup.ts`. Do not modify this file manually.
+
+// Flags: --no-warnings
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+
+const {
+  arrayBuffer,
+  blob,
+  buffer,
+  text,
+  json,
+} = require('stream/consumers');
+
+const {
+  Readable,
+  PassThrough
+} = require('stream');
+
+const {
+  TransformStream,
+} = require('stream/web');
+
+const buf = Buffer.from('hellothere');
+const kArrayBuffer =
+  buf.buffer.slice(buf.byteOffset, buf.byteOffset + buf.byteLength);
+
+{
+  const passthrough = new PassThrough();
+
+  blob(passthrough).then(common.mustCall(async (blob) => {
+    assert.strictEqual(blob.size, 10);
+    assert.deepStrictEqual(await blob.arrayBuffer(), kArrayBuffer);
+  }));
+
+  passthrough.write('hello');
+  setTimeout(() => passthrough.end('there'), 10);
+}
+
+{
+  const passthrough = new PassThrough();
+
+  arrayBuffer(passthrough).then(common.mustCall(async (ab) => {
+    assert.strictEqual(ab.byteLength, 10);
+    assert.deepStrictEqual(ab, kArrayBuffer);
+  }));
+
+  passthrough.write('hello');
+  setTimeout(() => passthrough.end('there'), 10);
+}
+
+{
+  const passthrough = new PassThrough();
+
+  buffer(passthrough).then(common.mustCall(async (buf) => {
+    assert.strictEqual(buf.byteLength, 10);
+    assert.deepStrictEqual(buf.buffer, kArrayBuffer);
+  }));
+
+  passthrough.write('hello');
+  setTimeout(() => passthrough.end('there'), 10);
+}
+
+
+{
+  const passthrough = new PassThrough();
+
+  text(passthrough).then(common.mustCall(async (str) => {
+    assert.strictEqual(str.length, 10);
+    assert.strictEqual(str, 'hellothere');
+  }));
+
+  passthrough.write('hello');
+  setTimeout(() => passthrough.end('there'), 10);
+}
+
+{
+  const readable = new Readable({
+    read() {}
+  });
+
+  text(readable).then((data) => {
+    assert.strictEqual(data, 'foo\ufffd\ufffd\ufffd');
+  });
+
+  readable.push(new Uint8Array([0x66, 0x6f, 0x6f, 0xed, 0xa0, 0x80]));
+  readable.push(null);
+}
+
+{
+  const passthrough = new PassThrough();
+
+  json(passthrough).then(common.mustCall(async (str) => {
+    assert.strictEqual(str.length, 10);
+    assert.strictEqual(str, 'hellothere');
+  }));
+
+  passthrough.write('"hello');
+  setTimeout(() => passthrough.end('there"'), 10);
+}
+
+{
+  const { writable, readable } = new TransformStream();
+
+  blob(readable).then(common.mustCall(async (blob) => {
+    assert.strictEqual(blob.size, 10);
+    assert.deepStrictEqual(await blob.arrayBuffer(), kArrayBuffer);
+  }));
+
+  const writer = writable.getWriter();
+  writer.write('hello');
+  setTimeout(() => {
+    writer.write('there');
+    writer.close();
+  }, 10);
+
+  assert.rejects(blob(readable), { code: 'ERR_INVALID_STATE' }).then(common.mustCall());
+}
+
+{
+  const { writable, readable } = new TransformStream();
+
+  arrayBuffer(readable).then(common.mustCall(async (ab) => {
+    assert.strictEqual(ab.byteLength, 10);
+    assert.deepStrictEqual(ab, kArrayBuffer);
+  }));
+
+  const writer = writable.getWriter();
+  writer.write('hello');
+  setTimeout(() => {
+    writer.write('there');
+    writer.close();
+  }, 10);
+
+  assert.rejects(arrayBuffer(readable), { code: 'ERR_INVALID_STATE' }).then(common.mustCall());
+}
+
+{
+  const { writable, readable } = new TransformStream();
+
+  text(readable).then(common.mustCall(async (str) => {
+    assert.strictEqual(str.length, 10);
+    assert.strictEqual(str, 'hellothere');
+  }));
+
+  const writer = writable.getWriter();
+  writer.write('hello');
+  setTimeout(() => {
+    writer.write('there');
+    writer.close();
+  }, 10);
+
+  assert.rejects(text(readable), { code: 'ERR_INVALID_STATE' }).then(common.mustCall());
+}
+
+{
+  const { writable, readable } = new TransformStream();
+
+  json(readable).then(common.mustCall(async (str) => {
+    assert.strictEqual(str.length, 10);
+    assert.strictEqual(str, 'hellothere');
+  }));
+
+  const writer = writable.getWriter();
+  writer.write('"hello');
+  setTimeout(() => {
+    writer.write('there"');
+    writer.close();
+  }, 10);
+
+  assert.rejects(json(readable), { code: 'ERR_INVALID_STATE' }).then(common.mustCall());
+}
+
+{
+  const stream = new PassThrough({
+    readableObjectMode: true,
+    writableObjectMode: true,
+  });
+
+  blob(stream).then(common.mustCall((blob) => {
+    assert.strictEqual(blob.size, 30);
+  }));
+
+  stream.write({});
+  stream.end({});
+}
+
+{
+  const stream = new PassThrough({
+    readableObjectMode: true,
+    writableObjectMode: true,
+  });
+
+  arrayBuffer(stream).then(common.mustCall((ab) => {
+    assert.strictEqual(ab.byteLength, 30);
+    assert.strictEqual(
+      Buffer.from(ab).toString(),
+      '[object Object][object Object]');
+  }));
+
+  stream.write({});
+  stream.end({});
+}
+
+{
+  const stream = new PassThrough({
+    readableObjectMode: true,
+    writableObjectMode: true,
+  });
+
+  buffer(stream).then(common.mustCall((buf) => {
+    assert.strictEqual(buf.byteLength, 30);
+    assert.strictEqual(
+      buf.toString(),
+      '[object Object][object Object]');
+  }));
+
+  stream.write({});
+  stream.end({});
+}
+
+{
+  const stream = new PassThrough({
+    readableObjectMode: true,
+    writableObjectMode: true,
+  });
+
+  assert.rejects(text(stream), {
+    code: 'ERR_INVALID_ARG_TYPE',
+  }).then(common.mustCall());
+
+  stream.write({});
+  stream.end({});
+}
+
+{
+  const stream = new PassThrough({
+    readableObjectMode: true,
+    writableObjectMode: true,
+  });
+
+  assert.rejects(json(stream), {
+    code: 'ERR_INVALID_ARG_TYPE',
+  }).then(common.mustCall());
+
+  stream.write({});
+  stream.end({});
+}
+
+{
+  const stream = new TransformStream();
+  text(stream.readable).then(common.mustCall((str) => {
+    // Incomplete utf8 character is flushed as a replacement char
+    assert.strictEqual(str.charCodeAt(0), 0xfffd);
+  }));
+  const writer = stream.writable.getWriter();
+  Promise.all([
+    writer.write(new Uint8Array([0xe2])),
+    writer.write(new Uint8Array([0x82])),
+    writer.close(),
+  ]).then(common.mustCall());
+}