stream: reject nested async streamables in from()

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,28 +277,37 @@ 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;
   }
 
   // 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,15 +316,15 @@ async function* normalizeAsyncValue(value) {
   // have both)
   if (isAsyncIterable(value)) {
     for await (const item of value) {
-      yield* normalizeAsyncValue(item);
+      yield* normalizeAsyncValue(item, allowNestedAsyncStreamables);
     }
     return;
   }
 
   // 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;

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(),