diff --git a/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp b/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp index 67990a061c7f..b832a9a759c5 100644 --- a/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp +++ b/src/jsc/bindings/node/JSNodeHTTPServerSocket.cpp @@ -1092,6 +1092,20 @@ extern "C" void Bun__NodeHTTP__onReadParsed(int ssl, us_socket_t* socket) } } +// Parsing stops first: a TLS close waits for unsent ciphertext, and uWS answers a request that it reads meanwhile. +extern "C" void Bun__NodeHTTP__refuseConnection(int ssl, us_socket_t* socket) +{ + if (us_socket_is_closed(socket)) { + return; + } + auto* serverSocket = ssl ? getNodeHTTPServerSocket(socket) : getNodeHTTPServerSocket(socket); + if (!serverSocket) { + return; + } + serverSocket->stopHTTPParsing(); + serverSocket->close(); +} + // Returns the JSNodeHTTPServerSocket already attached to this raw socket, or // creates (and attaches) one. Used for connections that have not produced a // parsed request yet: the 'clientError' path and the connection-accept path. diff --git a/src/runtime/server/server_body.rs b/src/runtime/server/server_body.rs index 36427494da12..75eca246851c 100644 --- a/src/runtime/server/server_body.rs +++ b/src/runtime/server/server_body.rs @@ -3288,15 +3288,18 @@ where /// its handshake completed). Hands the JSNodeHTTPServerSocket to the JS /// `onConnection` callback so `node:http` can emit 'connection' before any /// request bytes arrive. - pub(crate) fn on_connection_callback(&self, socket: *mut c_void) -> JsResult<()> { - if self.js_value_for_dispatch().is_none() { - return Ok(()); - } - let callback = self.on_connection; + pub(crate) fn on_connection_callback(this: *const Self, socket: *mut c_void) -> JsResult<()> { + let (callback, global) = { + // SAFETY: uWS runs the filter of a live server. Its listeners can free it, so this borrow ends before them. + let this = unsafe { &*this }; + if this.js_value_for_dispatch().is_none() { + return Ok(()); + } + (this.on_connection, this.global()) + }; if callback.is_empty() { return Ok(()); } - let global = self.global(); let node_socket = jsc::from_js_host_call(&global, || { Bun__getOrCreateNodeHTTPServerSocket(SSL, socket, &global) })?; @@ -3310,6 +3313,24 @@ where Ok(()) } + /// Node.js exits at a listener's error that nothing handles, and Bun goes on to read the request: https://github.com/nodejs/node/blob/v26.3.0/src/node_errors.cc#L1369-L1375 + #[inline(never)] // The filter thunk returns before this frame for 3 of its 4 calls per connection. + pub(crate) fn run_connection_listeners(this: *const Self, socket: *mut c_void) { + // SAFETY: uWS runs the filter of a live server. Its listeners can free it: `bun test` runs the next test from inside the report of their error. + let vm = unsafe { (*this).vm }; + let unhandled_errors = vm.unhandled_error_counter; + crate::dispatch::fold(Self::on_connection_callback(this, socket)); + if vm.unhandled_error_counter != unhandled_errors { + bun_core::hint::cold(); + // `bun test` takes every error of its main VM, as a handler does. + let runner_took_it = vm.worker.is_none() + && jsc::virtual_machine::isBunTest.load(core::sync::atomic::Ordering::Relaxed); + if !runner_took_it { + Bun__NodeHTTP__refuseConnection(c_int::from(SSL), socket); + } + } + } + // `js_gc_route_list_set` / `ptr_to_js` live on the unbounded // `impl NewServer` in mod.rs; do not redefine them here. } @@ -3457,9 +3478,10 @@ fn server_set_on_connection( } // SAFETY: user_data is the `*mut Self` registered below; // socket is a live uWS socket for this server's group. - // Shared — the callback re-enters JS. - let this = unsafe { &*user_data.cast::<$T>() }; - crate::dispatch::fold(this.on_connection_callback(socket.cast::())); + <$T>::run_connection_listeners( + user_data.cast::<$T>(), + socket.cast::(), + ); } if first { // S008: `NewApp` is a ZST opaque — safe `*mut → &mut` deref. @@ -3684,6 +3706,9 @@ unsafe extern "C" { global: &JSGlobalObject, ) -> JSValue; + // `socket` is the same live `uws::Socket*`: stops its HTTP parsing and closes it. + safe fn Bun__NodeHTTP__refuseConnection(ssl: c_int, socket: *mut c_void); + // `&JSGlobalObject` / `&mut JSValue` discharge the deref'd-param // preconditions; `request_ptr`/`req` are opaque handles that C++ stores or // forwards (module-private — sole callers pass live pointers). diff --git a/test/js/node/http/node-http-connection-listener-throw-fixture.js b/test/js/node/http/node-http-connection-listener-throw-fixture.js new file mode 100644 index 000000000000..72dc7f9063d9 --- /dev/null +++ b/test/js/node/http/node-http-connection-listener-throw-fixture.js @@ -0,0 +1,182 @@ +// A node:http server whose 'connection' or 'secureConnection' listener fails, and the peers of +// that server. The peers live in this process (the driver), which outlives the server: it prints +// what each peer read, how often the request listener ran and how the server ended. +// +// node-http-connection-listener-throw-fixture.js [worker:|hot:|watch:] +// +// A server case is a list of words: http or https, then what the listener and the process do. +// Runs unchanged under Node.js, except with --hot and --watch. The https cases read TLS_FIXTURE, +// the JSON of { cert, key }. +const { spawn } = require("node:child_process"); +const fs = require("node:fs"); +const http = require("node:http"); +const https = require("node:https"); +const net = require("node:net"); +const tls = require("node:tls"); +const { Worker, isMainThread, parentPort, workerData } = require("node:worker_threads"); + +// Neither process outlives a test that gave up on it. +let onDeadline = () => process.exit(124); +setTimeout(() => onDeadline(), 60_000).unref(); + +const REQUEST = "GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n"; +const REFUSAL = "HTTP/1.1 403 Forbidden\r\nContent-Length: 0\r\n\r\n"; +const [mode, serverCase] = splitCase(process.env.LISTENER_THROW_SERVER ?? ""); + +if (!isMainThread) server(workerData, port => parentPort.postMessage(port)); +else if (mode === "worker") serverInWorker(serverCase); +else if (serverCase) server(serverCase, port => console.log("PORT " + port)); +else driver(process.argv[2]); + +function splitCase(name) { + return name.includes(":") ? name.split(":") : ["run", name]; +} + +function server(kase, announce) { + const has = word => kase.split("-").includes(word); + const handled = has("handled") || has("captured"); + const page = (req, res) => { + console.log("REQUEST"); + res.end("SECRET", () => handled && srv.close()); + }; + const secure = has("https") && { + ...JSON.parse(process.env.TLS_FIXTURE), + // With TLS 1.2 the request is not in the read that ends the handshake. + ...(has("tls12") && { maxVersion: "TLSv1.2" }), + }; + const srv = secure ? https.createServer(secure, page) : http.createServer(page); + + const fail = () => { + throw new Error("PEER REFUSED"); + }; + srv.on(has("secureConnection") ? "secureConnection" : "connection", socket => { + // With --hot and --watch the process goes on after the error. It says so, and the driver stops it. + if (process.env.LISTENER_THROW_ALIVE) setImmediate(() => console.log("ALIVE")); + if (has("destroys")) return socket.on("close", () => srv.close()).destroy(); + // "big" is more than the socket takes at once. The "unread" peer takes nothing of it, and + // the close of a TLS socket then waits for what is not sent, with the request behind it. + if (has("writes")) socket.write(has("big") ? Buffer.alloc(4 << 20, "-") : REFUSAL); + if (has("destroyed")) socket.destroy(); + if (has("ends")) socket.end(REFUSAL); + // An immediate keeps the loop turning after the error. + if (has("immediate")) setImmediate(() => {}); + if (has("nextTick")) return process.nextTick(fail); + if (has("queueMicrotask")) return queueMicrotask(fail); + fail(); + }); + + if (has("monitor")) process.on("uncaughtExceptionMonitor", () => {}); + if (has("handled")) process.on("uncaughtException", () => {}); + if (has("captured")) process.setUncaughtExceptionCaptureCallback(() => {}); + if (has("exits")) process.on("uncaughtException", () => process.exit(3)); + + srv.listen(0, "127.0.0.1", () => { + announce(srv.address().port); + // Hold this turn of the loop until the driver has its peers in the accept queue: the next + // turn finds all of them. The read ends with nothing when the driver is gone. + if (fs.readSync(0, Buffer.alloc(1)) === 0) process.exit(125); + }); +} + +function serverInWorker(kase) { + const worker = new Worker(__filename, { workerData: kase }); + worker.on("message", port => console.log("PORT " + port)); + // The error that ends the Worker. The process ends behind it, with 0. + worker.on("error", () => {}); +} + +async function driver(name) { + const [mode, kase] = splitCase(name); + const has = word => kase.split("-").includes(word); + const watching = mode === "hot" || mode === "watch"; + const child = spawn(process.execPath, watching ? ["--" + mode, __filename] : [__filename], { + env: { ...process.env, LISTENER_THROW_SERVER: name, ...(watching && { LISTENER_THROW_ALIVE: "1" }) }, + stdio: ["pipe", "pipe", "pipe"], + }); + process.on("exit", () => child.kill()); + let stdout = ""; + let stderr = ""; + child.stdout.setEncoding("utf8").on("data", chunk => (stdout += chunk)); + child.stderr.setEncoding("utf8").on("data", chunk => (stderr += chunk)); + const exited = new Promise(resolve => child.on("close", (code, signal) => resolve(signal ?? code))); + // When the driver cannot go on, the test gets what the server printed. + const giveUp = why => { + process.stderr.write(`${name}: ${why}\n--- stdout of the server\n${stdout}\n--- stderr of the server\n${stderr}\n`); + process.exit(1); + }; + onDeadline = () => giveUp("the case did not finish"); + const printed = async line => { + while (!line.test(stdout)) { + const more = new Promise(resolve => child.stdout.once("data", resolve)); + const ended = await Promise.race([more, exited.then(() => true)]); + if (ended === true && !line.test(stdout)) giveUp(`the server ended before it printed ${line}`); + } + return line.exec(stdout); + }; + const port = +(await printed(/^PORT (\d+)$/m))[1]; + + const secure = has("https"); + // "unread": the peer reads nothing until the server is done with it. Only a socket with + // onread stops its reads at once when it is paused. + const unread = has("unread"); + const peers = Array.from({ length: name === "http" ? 2 : 1 }, () => { + const peer = { read: "" }; + const onread = { + buffer: Buffer.alloc(1 << 16), + callback: (length, buffer) => { + peer.read += buffer.latin1Slice(0, length); + return true; + }, + }; + const socket = secure + ? tls.connect( + { + host: "127.0.0.1", + port, + ca: JSON.parse(process.env.TLS_FIXTURE).cert, + servername: "localhost", + ...(unread && { onread }), + }, + () => { + if (unread) socket.pause(); + socket.write(REQUEST); + }, + ) + : net.connect(port, "127.0.0.1"); + // In the accept queue: the TCP handshake is done and, without TLS, the request is sent. + peer.queued = new Promise(resolve => + socket.once("connect", () => { + peer.connected = true; + if (secure) resolve(); + else socket.write(REQUEST, resolve); + }), + ); + peer.closed = new Promise(resolve => socket.on("close", resolve)); + if (!unread) socket.setEncoding("utf8").on("data", chunk => (peer.read += chunk)); + // After the connect, an error is how a refused peer sees the close of the server. + socket.on("error", error => peer.connected || giveUp("a peer did not connect: " + error.code)); + peer.socket = socket; + return peer; + }); + await Promise.all(peers.map(peer => peer.queued)); + child.stdin.end("go"); + if (unread) { + await (watching ? printed(/^ALIVE$/m) : exited); + for (const peer of peers) peer.socket.resume(); + } + await Promise.all(peers.map(peer => peer.closed)); + if (watching) { + await printed(/^ALIVE$/m); + child.kill(); + } + + const exit = await exited; + console.log( + JSON.stringify({ + requests: stdout.split("\n").filter(line => line === "REQUEST").length, + served: peers.filter(peer => peer.read.includes("SECRET")).length, + read: peers.map(peer => peer.read.length), + exit, + }), + ); +} diff --git a/test/js/node/http/node-http-connection-listener-throw.test.ts b/test/js/node/http/node-http-connection-listener-throw.test.ts new file mode 100644 index 000000000000..8c62890c2690 --- /dev/null +++ b/test/js/node/http/node-http-connection-listener-throw.test.ts @@ -0,0 +1,167 @@ +import { describe, expect, test } from "bun:test"; +import { bunEnv, bunExe, isASAN, isCI, isDebug, isWindows, tempDir, tls as tlsCert } from "harness"; +import { join } from "node:path"; + +// When nothing handles the error of a 'connection' or 'secureConnection' listener, Node.js exits +// at the throw. Bun exits after the turn of the loop, and does not answer that connection before +// it. When something handles the error, the connection stays and its request is answered. +// +// The fixture is the server and its peers. It runs unchanged under Node.js. +describe.concurrent("an http server whose connection listener throws", () => { + const fixture = join(import.meta.dir, "node-http-connection-listener-throw-fixture.js"); + // CI has its own time for each test. A local debug or ASAN build needs several seconds to start one process. + const timeout = (isASAN || isDebug) && !isCI ? 90_000 : undefined; + // A Worker with a server leaks the request pool of its thread (#37513). The stack of that + // allocation reaches a frame that test/leaksan.supp names only beyond the 30 frames of CI. + const workerEnv = { LSAN_OPTIONS: [bunEnv.LSAN_OPTIONS, "malloc_context_size=100"].filter(Boolean).join(":") }; + + const refused = { requests: 0, served: 0, read: [0], exit: 1 }; + const answered = { requests: 1, served: 1, read: [100], exit: 0 }; + const expected: Record = { + // What Node.js v26.3.0 prints for the same case. + // Two peers wait in the accept queue, with their requests sent. + "http": { ...refused, read: [0, 0] }, + // Node.js emits 'connection' before the TLS handshake, Bun after it. The peer reads the same. + "https": refused, + "https-secureConnection": refused, + "http-nextTick": refused, + "http-queueMicrotask": refused, + "http-monitor": refused, + // The loop has more to do after the error, so the request is read without a deferred accept too. + "http-immediate": refused, + // The listener writes a 403 before it throws, and the peer reads that. The close behind it is + // a reset, because the request is not read, and on Windows a reset can discard the 403. + "http-writes": { ...refused, read: [isWindows ? expect.any(Number) : 45] }, + // The close waits for what the peer has not taken, and the request behind it is not answered. + "https-secureConnection-writes-big-unread": { ...refused, read: [expect.any(Number)] }, + "http-handled": answered, + "https-secureConnection-handled": answered, + "http-captured": answered, + "http-exits": { ...refused, exit: 3 }, + // The listener refuses with socket.destroy() and throws nothing. + "http-destroys": { ...refused, exit: 0 }, + // The listener closes the socket itself before it throws. + "http-destroyed": refused, + "https-secureConnection-destroyed": refused, + "http-ends": { ...refused, read: [isWindows ? expect.any(Number) : 45] }, + "worker:http": { ...refused, exit: 0 }, + "worker:http-nextTick": { ...refused, exit: 0 }, + + // Not what Node.js prints. Its Worker ends behind the read that holds the request of this + // peer, and answers it. Without TLS 1.3 it does not. + "worker:https-secureConnection": { ...refused, exit: 0 }, + // The process stays up, and the driver stops it. + "hot:http": { ...refused, exit: "SIGTERM" }, + "hot:https-secureConnection": { ...refused, exit: "SIGTERM" }, + // The request comes in a later read than the end of the handshake. + "hot:https-secureConnection-tls12": { ...refused, exit: "SIGTERM" }, + "watch:http": { ...refused, exit: "SIGTERM" }, + }; + + test.each(Object.keys(expected))( + "%s", + async name => { + await using proc = Bun.spawn({ + cmd: [bunExe(), fixture, name], + env: { ...bunEnv, TLS_FIXTURE: JSON.stringify(tlsCert), ...(name.startsWith("worker:") && workerEnv) }, + stdout: "pipe", + stderr: "pipe", + stdin: "ignore", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ + result: stdout ? JSON.parse(stdout) : undefined, + stderr, + exitCode, + signalCode: proc.signalCode, + }).toEqual({ + result: expected[name], + stderr: "", + exitCode: 0, + signalCode: null, + }); + }, + timeout, + ); + + // `bun test` takes the error as the failure of the test that runs, so the connection stays. + // It starts the next test from inside the report of the error. When that test keeps the loop + // to itself, the server of the failed test is closed, collected and freed before the report + // returns to the frame that ran the listener. + test( + "under bun test", + async () => { + using dir = tempDir("connection-listener-throw", { + "listener.test.js": ` + import { expect, test } from "bun:test"; + import http from "node:http"; + import net from "node:net"; + + // A server whose connection listener throws, and one peer of it. + async function refusedPeer() { + const server = http.createServer((req, res) => res.end("SECRET")); + server.on("connection", () => { + throw new Error("PEER REFUSED"); + }); + await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); + let read = ""; + await new Promise(resolve => { + const peer = net.connect(server.address().port, "127.0.0.1", () => + peer.write("GET / HTTP/1.1\\r\\nHost: localhost\\r\\nConnection: close\\r\\n\\r\\n"), + ); + peer.setEncoding("utf8"); + peer.on("data", chunk => (read += chunk)); + peer.on("error", () => {}); + peer.on("close", resolve); + }); + console.log("READ " + read.length); + await new Promise(resolve => server.close(resolve)); + } + + let first; + let second; + const blocks = []; + test("a connection listener throws", () => (first = refusedPeer())); + test("the next test gives the loop back", () => first); + // Not on Windows: the wait below runs inside the open callback of the connection that it waits for, and does not always end there (#33261). + const posix = test.skipIf(process.platform === "win32"); + posix("a connection listener throws again", () => (second = refusedPeer())); + posix("the next test keeps the loop", () => { + expect( + (async () => { + await second; + Bun.gc(true); + await new Promise(resolve => setImmediate(resolve)); + await new Promise(resolve => setImmediate(resolve)); + // Blocks of the size of the freed server: a read of it finds them. + for (let i = 0; i < 4096; i++) blocks.push(Bun.allocUnsafe(1700).fill(0x41)); + return blocks.length; + })(), + ).resolves.toBe(4096); + }); + `, + }); + await using proc = Bun.spawn({ + cmd: [bunExe(), "test", "./listener.test.js"], + env: bunEnv, + cwd: String(dir), + stdout: "pipe", + stderr: "pipe", + stdin: "ignore", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ + read: stdout.match(/^READ \d+$/gm), + summary: stderr.match(/^ *\d+ (?:pass|fail)$/gm)?.map(line => line.trim()), + exitCode, + signalCode: proc.signalCode, + }).toEqual({ + read: isWindows ? ["READ 100"] : ["READ 100", "READ 100"], + summary: isWindows ? ["1 pass", "1 fail"] : ["2 pass", "2 fail"], + exitCode: 1, + signalCode: null, + }); + }, + timeout, + ); +});