From 2e3bde5e3dc80658f8b5d3dbd3fd8c0315a27367 Mon Sep 17 00:00:00 2001
From: "Kamat, Trivikram" <16024985+trivikr@users.noreply.github.com>
Date: Tue, 7 Jul 2026 23:34:04 -0700
Subject: [PATCH] stream: reject nested async streamables in from()

Reject nested async iterables and toAsyncStreamable values when from()
is consuming a sync iterable source. Promise values are still awaited.

Signed-off-by: Kamat, Trivikram <16024985+trivikr@users.noreply.github.com>
Assisted-by: openai:gpt-5.5
---
 lib/internal/streams/iter/from.js            | 27 +++++++++++------
 test/parallel/test-stream-iter-from-async.js | 32 ++++++++++++++++++++
 2 files changed, 50 insertions(+), 9 deletions(-)

diff --git a/lib/internal/streams/iter/from.js b/lib/internal/streams/iter/from.js
index 1efe83e9a04162..194bf40d691d12 100644
--- a/lib/internal/streams/iter/from.js
+++ b/lib/internal/streams/iter/from.js
@@ -263,11 +263,11 @@ function* normalizeSyncSource(source) {
  * and protocol conversions.
  * @yields {Uint8Array}
  */
-async function* normalizeAsyncValue(value) {
+async function* normalizeAsyncValue(value, allowNestedAsyncStreamables = true) {
   // Handle promises first
   if (isPromise(value)) {
     const resolved = await value;
-    yield* normalizeAsyncValue(resolved);
+    yield* normalizeAsyncValue(resolved, allowNestedAsyncStreamables);
     return;
   }
 
@@ -277,13 +277,22 @@ async function* normalizeAsyncValue(value) {
     return;
   }
 
+  if (!allowNestedAsyncStreamables &&
+      (isAsyncIterable(value) || hasProtocol(value, toAsyncStreamable))) {
+    throw new ERR_INVALID_ARG_TYPE(
+      'value',
+      ['string', 'ArrayBuffer', 'ArrayBufferView', 'Iterable', 'toStreamable'],
+      value,
+    );
+  }
+
   // Handle ToAsyncStreamable protocol (check before ToStreamable)
   if (hasProtocol(value, toAsyncStreamable)) {
     const result = FunctionPrototypeCall(value[toAsyncStreamable], value);
     if (isPromise(result)) {
-      yield* normalizeAsyncValue(await result);
+      yield* normalizeAsyncValue(await result, allowNestedAsyncStreamables);
     } else {
-      yield* normalizeAsyncValue(result);
+      yield* normalizeAsyncValue(result, allowNestedAsyncStreamables);
     }
     return;
   }
@@ -291,14 +300,14 @@ async function* normalizeAsyncValue(value) {
   // Handle ToStreamable protocol
   if (hasProtocol(value, toStreamable)) {
     const result = FunctionPrototypeCall(value[toStreamable], value);
-    yield* normalizeAsyncValue(result);
+    yield* normalizeAsyncValue(result, allowNestedAsyncStreamables);
     return;
   }
 
   // Handle arrays (which are also iterable, but check first for efficiency)
   if (ArrayIsArray(value)) {
     for (let i = 0; i < value.length; i++) {
-      yield* normalizeAsyncValue(value[i]);
+      yield* normalizeAsyncValue(value[i], allowNestedAsyncStreamables);
     }
     return;
   }
@@ -307,7 +316,7 @@ async function* normalizeAsyncValue(value) {
   // have both)
   if (isAsyncIterable(value)) {
     for await (const item of value) {
-      yield* normalizeAsyncValue(item);
+      yield* normalizeAsyncValue(item, allowNestedAsyncStreamables);
     }
     return;
   }
@@ -315,7 +324,7 @@ async function* normalizeAsyncValue(value) {
   // Handle sync iterables
   if (isSyncIterable(value)) {
     for (const item of value) {
-      yield* normalizeAsyncValue(item);
+      yield* normalizeAsyncValue(item, allowNestedAsyncStreamables);
     }
     return;
   }
@@ -392,7 +401,7 @@ async function* normalizeAsyncSource(source) {
         batch = [];
       }
       let asyncBatch = [];
-      for await (const chunk of normalizeAsyncValue(value)) {
+      for await (const chunk of normalizeAsyncValue(value, false)) {
         ArrayPrototypePush(asyncBatch, chunk);
         if (asyncBatch.length === FROM_BATCH_SIZE) {
           yield asyncBatch;
diff --git a/test/parallel/test-stream-iter-from-async.js b/test/parallel/test-stream-iter-from-async.js
index 8080b7a5cd86ca..5ef78088bfbd4c 100644
--- a/test/parallel/test-stream-iter-from-async.js
+++ b/test/parallel/test-stream-iter-from-async.js
@@ -49,6 +49,35 @@ async function testFromSyncIterableAsAsync() {
   assert.deepStrictEqual(batches[0][1], new Uint8Array([2]));
 }
 
+async function testFromSyncIterableAwaitsPromiseValues() {
+  const result = await text(from([Promise.resolve('promise-value')]));
+  assert.strictEqual(result, 'promise-value');
+}
+
+async function testFromSyncIterableRejectsNestedAsyncIterable() {
+  async function* asyncGenerator() {
+    yield 'data';
+  }
+
+  await assert.rejects(
+    () => text(from([asyncGenerator()])),
+    { code: 'ERR_INVALID_ARG_TYPE' },
+  );
+}
+
+async function testFromSyncIterableRejectsNestedToAsyncStreamable() {
+  const obj = {
+    [Symbol.for('Stream.toAsyncStreamable')]() {
+      return 'data';
+    },
+  };
+
+  await assert.rejects(
+    () => text(from([obj])),
+    { code: 'ERR_INVALID_ARG_TYPE' },
+  );
+}
+
 async function testFromToAsyncStreamableProtocol() {
   const sym = Symbol.for('Stream.toAsyncStreamable');
   const obj = {
@@ -232,6 +261,9 @@ Promise.all([
   testFromString(),
   testFromAsyncGenerator(),
   testFromSyncIterableAsAsync(),
+  testFromSyncIterableAwaitsPromiseValues(),
+  testFromSyncIterableRejectsNestedAsyncIterable(),
+  testFromSyncIterableRejectsNestedToAsyncStreamable(),
   testFromToAsyncStreamableProtocol(),
   testFromRejectsNonStreamable(),
   testFromEmptyArray(),