fix(node): implement TLSSocket._start (denoland#20120)

Closes denoland#19983
Closes denoland#18303
Closes denoland#16681
Closes denoland#19978

cli/tests/unit_node/tls_test.ts

@@ -56,6 +56,38 @@ Connection: close
  await serve;
 });
 
+// https://github.com/denoland/deno/pull/20120
+Deno.test("tls.connect mid-read tcp->tls upgrade", async () => {
+ const ctl = new AbortController();
+ const serve = serveTls(() => new Response("hello"), {
+ port: 8443,
+ key,
+ cert,
+ signal: ctl.signal,
+ });
+
+ await delay(200);
+
+ const conn = tls.connect({
+ host: "localhost",
+ port: 8443,
+ secureContext: {
+ ca: rootCaCert,
+ // deno-lint-ignore no-explicit-any
+ } as any,
+ });
+
+ conn.setEncoding("utf8");
+ conn.write(`GET / HTTP/1.1\nHost: www.google.com\n\n`);
+
+ conn.on("data", (_) => {
+ conn.destroy();
+ ctl.abort();
+ });
+
+ await serve;
+});
+
 Deno.test("tls.createServer creates a TLS server", async () => {
  const p = deferred();
  const server = tls.createServer(

ext/node/polyfills/_tls_wrap.ts

@@ -26,6 +26,11 @@ import {
 import { EventEmitter } from "node:events";
 import { kEmptyObject } from "ext:deno_node/internal/util.mjs";
 import { nextTick } from "ext:deno_node/_next_tick.ts";
+import { kHandle } from "ext:deno_node/internal/stream_base_commons.ts";
+import {
+ isAnyArrayBuffer,
+ isArrayBufferView,
+} from "ext:deno_node/internal/util/types.ts";
 
 const kConnectOptions = Symbol("connect-options");
 const kIsVerified = Symbol("verified");
@@ -71,7 +76,11 @@ export class TLSSocket extends net.Socket {
  [kPendingSession]: any;
  [kConnectOptions]: any;
  ssl: any;
- _start: any;
+
+ _start() {
+ this[kHandle].afterConnect();
+ }
+
  constructor(socket: any, opts: any = kEmptyObject) {
  const tlsOptions = { ...opts };
 
@@ -84,6 +93,9 @@ export class TLSSocket extends net.Socket {
 
  let caCerts = tlsOptions?.secureContext?.ca;
  if (typeof caCerts === "string") caCerts = [caCerts];
+ else if (isArrayBufferView(caCerts) || isAnyArrayBuffer(caCerts)) {
+ caCerts = [new TextDecoder().decode(caCerts)];
+ }
  tlsOptions.caCerts = caCerts;
 
  super({
@@ -139,9 +151,9 @@ export class TLSSocket extends net.Socket {
  handle.afterConnect = async (req: any, status: number) => {
  try {
  const conn = await Deno.startTls(handle[kStreamBaseField], options);
+ handle[kStreamBaseField] = conn;
  tlssock.emit("secure");
  tlssock.removeListener("end", onConnectEnd);
- handle[kStreamBaseField] = conn;
  } catch {
  // TODO(kt3k): Handle this
  }

ext/node/polyfills/internal_binding/stream_wrap.ts

@@ -314,9 +314,16 @@ export class LibuvStreamWrap extends HandleWrap {
  let buf = BUF;
 
  let nread: number | null;
+ const ridBefore = this[kStreamBaseField]!.rid;
  try {
  nread = await this[kStreamBaseField]!.read(buf);
  } catch (e) {
+ // Try to read again if the underlying stream resource
+ // changed. This can happen during TLS upgrades (eg. STARTTLS)
+ if (ridBefore != this[kStreamBaseField]!.rid) {
+ return this.#read();
+ }
+
  if (
  e instanceof Deno.errors.Interrupted ||
  e instanceof Deno.errors.BadResource
@@ -365,15 +372,23 @@ export class LibuvStreamWrap extends HandleWrap {
  async #write(req: WriteWrap<LibuvStreamWrap>, data: Uint8Array) {
  const { byteLength } = data;
 
+ const ridBefore = this[kStreamBaseField]!.rid;
+
+ let nwritten = 0;
  try {
  // TODO(crowlKats): duplicate from runtime/js/13_buffer.js
- let nwritten = 0;
  while (nwritten < data.length) {
  nwritten += await this[kStreamBaseField]!.write(
  data.subarray(nwritten),
  );
  }
  } catch (e) {
+ // Try to read again if the underlying stream resource
+ // changed. This can happen during TLS upgrades (eg. STARTTLS)
+ if (ridBefore != this[kStreamBaseField]!.rid) {
+ return this.#write(req, data.subarray(nwritten));
+ }
+
  let status: number;
 
  // TODO(cmorten): map err to status codes