feat(ext/net): use rustls_tokio_stream (#21205)

Fixes #21121 and #19498

Migrates fully to rustls_tokio_stream. We no longer need to maintain our
own TlsStream implementation to properly support duplex.

This should fix a number of errors with TLS and websockets, HTTP and
"other" places where it's failing.

1 files changed, 45 insertions, 21 deletions

diff --git a/cli/tests/unit/tls_test.ts b/cli/tests/unit/tls_test.ts
index 8162c53b5..31e24d547 100644
--- a/cli/tests/unit/tls_test.ts
+++ b/cli/tests/unit/tls_test.ts
@@ -11,6 +11,7 @@ import {
 } from "./test_util.ts";
 import { BufReader, BufWriter } from "../../../test_util/std/io/mod.ts";
 import { readAll } from "../../../test_util/std/streams/read_all.ts";
+import { writeAll } from "../../../test_util/std/streams/write_all.ts";
 import { TextProtoReader } from "../testdata/run/textproto.ts";
 
 const encoder = new TextEncoder();
@@ -538,15 +539,23 @@ Deno.test(
   },
 );
 
+const largeAmount = 1 << 20 /* 1 MB */;
+
 async function sendAlotReceiveNothing(conn: Deno.Conn) {
   // Start receive op.
   const readBuf = new Uint8Array(1024);
   const readPromise = conn.read(readBuf);
 
+  const timeout = setTimeout(() => {
+    throw new Error("Failed to send buffer in a reasonable amount of time");
+  }, 10_000);
+
   // Send 1 MB of data.
-  const writeBuf = new Uint8Array(1 << 20 /* 1 MB */);
+  const writeBuf = new Uint8Array(largeAmount);
   writeBuf.fill(42);
-  await conn.write(writeBuf);
+  await writeAll(conn, writeBuf);
+
+  clearTimeout(timeout);
 
   // Send EOF.
   await conn.closeWrite();
@@ -564,14 +573,29 @@ async function sendAlotReceiveNothing(conn: Deno.Conn) {
 async function receiveAlotSendNothing(conn: Deno.Conn) {
   const readBuf = new Uint8Array(1024);
   let n: number | null;
+  let nread = 0;
+
+  const timeout = setTimeout(() => {
+    throw new Error(
+      `Failed to read buffer in a reasonable amount of time (got ${nread}/${largeAmount})`,
+    );
+  }, 10_000);
 
   // Receive 1 MB of data.
-  for (let nread = 0; nread < 1 << 20 /* 1 MB */; nread += n!) {
-    n = await conn.read(readBuf);
-    assertStrictEquals(typeof n, "number");
-    assert(n! > 0);
-    assertStrictEquals(readBuf[0], 42);
+  try {
+    for (; nread < largeAmount; nread += n!) {
+      n = await conn.read(readBuf);
+      assertStrictEquals(typeof n, "number");
+      assert(n! > 0);
+      assertStrictEquals(readBuf[0], 42);
+    }
+  } catch (e) {
+    throw new Error(
+      `Got an error (${e.message}) after reading ${nread}/${largeAmount} bytes`,
+      { cause: e },
+    );
   }
+  clearTimeout(timeout);
 
   // Close the connection, without sending anything at all.
   conn.close();
@@ -623,7 +647,7 @@ async function sendReceiveEmptyBuf(conn: Deno.Conn) {
 
   await assertRejects(async () => {
     await conn.write(byteBuf);
-  }, Deno.errors.BrokenPipe);
+  }, Deno.errors.NotConnected);
 
   n = await conn.write(emptyBuf);
   assertStrictEquals(n, 0);
@@ -841,13 +865,12 @@ async function tlsWithTcpFailureTestImpl(
           tcpForwardingInterruptPromise2.resolve();
           break;
         case "shutdown":
-          // Receiving a TCP FIN packet without receiving a TLS CloseNotify
-          // alert is not the expected mode of operation, but it is not a
-          // problem either, so it should be treated as if the TLS session was
-          // gracefully closed.
           await Promise.all([
             tcpConn1.closeWrite(),
-            await receiveEof(tlsConn1),
+            await assertRejects(
+              () => receiveEof(tlsConn1),
+              Deno.errors.UnexpectedEof,
+            ),
             await tlsConn1.closeWrite(),
             await receiveEof(tlsConn2),
           ]);
@@ -1036,8 +1059,8 @@ function createHttpsListener(port: number): Deno.Listener {
       );
 
       // Send response.
-      await conn.write(resHead);
-      await conn.write(resBody);
+      await writeAll(conn, resHead);
+      await writeAll(conn, resBody);
 
       // Close TCP connection.
       conn.close();
@@ -1046,12 +1069,14 @@ function createHttpsListener(port: number): Deno.Listener {
 }
 
 async function curl(url: string): Promise<string> {
-  const { success, code, stdout } = await new Deno.Command("curl", {
+  const { success, code, stdout, stderr } = await new Deno.Command("curl", {
     args: ["--insecure", url],
   }).output();
 
   if (!success) {
-    throw new Error(`curl ${url} failed: ${code}`);
+    throw new Error(
+      `curl ${url} failed: ${code}:\n${new TextDecoder().decode(stderr)}`,
+    );
   }
   return new TextDecoder().decode(stdout);
 }
@@ -1276,8 +1301,7 @@ Deno.test(
     // Begin sending a 10mb blob over the TLS connection.
     const whole = new Uint8Array(10 << 20); // 10mb.
     whole.fill(42);
-    const sendPromise = conn1.write(whole);
-
+    const sendPromise = writeAll(conn1, whole);
     // Set up the other end to receive half of the large blob.
     const half = new Uint8Array(whole.byteLength / 2);
     const receivePromise = readFull(conn2, half);
@@ -1294,7 +1318,7 @@ Deno.test(
 
     // Receive second half of large blob. Wait for the send promise and check it.
     assertEquals(await readFull(conn2, half), half.length);
-    assertEquals(await sendPromise, whole.length);
+    await sendPromise;
 
     await conn1.handshake();
     await conn2.handshake();
@@ -1352,7 +1376,7 @@ Deno.test(
       await assertRejects(
         () => conn.handshake(),
         Deno.errors.InvalidData,
-        "UnknownIssuer",
+        "invalid peer certificate: UnknownIssuer",
       );
       conn.close();
     }