diff --git a/src/http_jsc/websocket_client/CppWebSocket.rs b/src/http_jsc/websocket_client/CppWebSocket.rs index cb3fec21054..e948fc82e35 100644 --- a/src/http_jsc/websocket_client/CppWebSocket.rs +++ b/src/http_jsc/websocket_client/CppWebSocket.rs @@ -22,6 +22,28 @@ bun_opaque::opaque_ffi! { pub struct CppWebSocket; } +/// A single handshake response header borrowed from the parse buffer, matching +/// the C++ `WebCore::WebSocket::HandshakeRawHeader` layout. Only valid for the +/// duration of the [`CppWebSocket::did_receive_handshake_response`] call. +#[repr(C)] +pub(crate) struct HandshakeRawHeader { + name_ptr: *const u8, + name_len: usize, + value_ptr: *const u8, + value_len: usize, +} + +impl HandshakeRawHeader { + pub(crate) fn new(name: &[u8], value: &[u8]) -> Self { + Self { + name_ptr: name.as_ptr(), + name_len: name.len(), + value_ptr: value.as_ptr(), + value_len: value.len(), + } + } +} + // FFI surface for `WebCore::WebSocket` (src/jsc/bindings/webcore/WebSocket.cpp). // Kept private to this module — the safe wrappers below are the only callers. // @@ -61,6 +83,16 @@ unsafe extern "C" { safe fn WebSocket__incrementPendingActivity(websocket_context: &CppWebSocket); safe fn WebSocket__decrementPendingActivity(websocket_context: &CppWebSocket); fn WebSocket__setProtocol(websocket_context: &CppWebSocket, protocol: *mut BunString); + fn WebSocket__didReceiveHandshakeResponse( + websocket_context: &CppWebSocket, + status_code: u16, + status_message: *const u8, + status_message_len: usize, + headers: *const HandshakeRawHeader, + headers_len: usize, + body: *const u8, + body_len: usize, + ); } // Receivers are `&self` (not `&mut self`) because `CppWebSocket` is @@ -173,6 +205,38 @@ impl CppWebSocket { }; event_loop.exit(); } + + /// Forward the parsed HTTP handshake response to C++ so the `ws` shim can + /// emit `upgrade` / `unexpected-response`. The C++ side skips all work when + /// no `handshake` listener is registered, so the common browser-style path + /// stays zero-cost. All slices must outlive the call (they borrow the parse + /// buffer, which the caller keeps alive). + pub(crate) fn did_receive_handshake_response( + &self, + status_code: u16, + status_message: &[u8], + headers: &[HandshakeRawHeader], + body: &[u8], + ) { + // SAFETY: VirtualMachine::get() returns the live current-thread VM; + // event_loop() yields its raw event-loop pointer (live for VM lifetime). + let event_loop = VirtualMachine::get().event_loop_mut(); + event_loop.enter(); + // SAFETY: self is a valid C++ WebCore::WebSocket; all slices outlive the call. + unsafe { + WebSocket__didReceiveHandshakeResponse( + self, + status_code, + status_message.as_ptr(), + status_message.len(), + headers.as_ptr(), + headers.len(), + body.as_ptr(), + body.len(), + ) + }; + event_loop.exit(); + } } impl CppWebSocket { diff --git a/src/http_jsc/websocket_client/WebSocketUpgradeClient.rs b/src/http_jsc/websocket_client/WebSocketUpgradeClient.rs index 8cdbbf6bbf7..c3b0074f152 100644 --- a/src/http_jsc/websocket_client/WebSocketUpgradeClient.rs +++ b/src/http_jsc/websocket_client/WebSocketUpgradeClient.rs @@ -37,7 +37,7 @@ use bun_picohttp as picohttp; use bun_ptr::ThisPtr; use bun_uws::{self as uws, SocketHandler, SocketKind, SslCtx}; -use super::cpp_websocket::CppWebSocket; +use super::cpp_websocket::{CppWebSocket, HandshakeRawHeader}; use super::websocket_deflate as WebSocketDeflate; use super::websocket_proxy::WebSocketProxy; use super::websocket_proxy_tunnel::WebSocketProxyTunnel; @@ -1561,17 +1561,74 @@ impl HTTPClient { return; } - // Ownership transfer: `overflow` is HANDED OFF across FFI — + // Forward the parsed 101 handshake response to the C++ WebSocket so the + // `ws` shim can emit `upgrade` (before `open`). The C++ side is a no-op + // unless a `handshake` listener is registered, so the browser-style + // `new WebSocket()` path pays nothing. This dispatches into JS *before* + // `did_connect` (while the socket is still CONNECTING, matching node's + // `ws`) so a `handshake`/`upgrade` handler that synchronously closes the + // socket is handled by the existing `!tcp.is_closed() && has_ws` + // re-check below: `cancel()` takes `outgoing_websocket`, so `has_ws` + // becomes false and `did_connect` is skipped. + // + // The header name/value slices borrow the parse buffer (`body` in + // `handle_data`), which is still alive here — `clear_data()` (which + // frees it) runs further below. The body is passed empty: for a 101 the + // trailing `remain_buf` bytes are the first WebSocket frame (not an HTTP + // body), and they're forwarded to the protocol reader via `did_connect` + // below — the `ws` shim drops the handshake event's body anyway, so + // don't copy them into a Uint8Array that's immediately discarded. (The + // C++ `body` param stays for a future `unexpected-response`, which would + // carry a real HTTP body.) + // + // Bracket the handshake dispatch AND the `did_connect`/ + // `did_connect_with_tunnel` handoff below in one event-loop scope. Each + // of those FFI wrappers does its own `enter()/exit()`, and the uWS poll + // that drives `handle_data` runs at `entered_event_loop_count == 0`, so + // without this outer scope the `exit()` after the `upgrade` dispatch + // would hit count 1 and drain microtasks / `process.nextTick` *before* + // `open` fires. Node + `ws` emit `upgrade` and `open` from the same + // socket-data turn with no checkpoint between them; the outer scope + // makes the inner pairs nest (count stays ≥ 1, no drain) so queued + // microtasks run after `open`, matching node. Drops at function end, + // after `open` and the trailing derefs (which may free `this`; the + // guard holds only the VM-owned loop pointer, not `this`). + let _event_loop_scope = + bun_jsc::virtual_machine::VirtualMachine::get().enter_event_loop_scope(); + + // SAFETY: short-lived read of `outgoing_websocket`. + if let Some(ws) = unsafe { (*this).outgoing_websocket } { + let mut raw_headers: Vec = + Vec::with_capacity(response.headers.list.len()); + for header in response.headers.list { + raw_headers.push(HandshakeRawHeader::new(header.name(), header.value())); + } + let status_code = u16::try_from(response.status_code).unwrap_or(0); + CppWebSocket::opaque_ref(ws).did_receive_handshake_response( + status_code, + response.status, + &raw_headers, + &[], + ); + } + + // Owned copy of the bytes that trailed the 101 header block. It is + // HANDED OFF across FFI in the success arms below — // `WebSocket__didConnect` → `Bun__WebSocketClient__init`/`_initWithTunnel` // adopts the raw `(ptr, len)` into an `InitialDataHandler` queued as a // microtask, which reclaims it via `Box::<[u8]>::from_raw` when the - // microtask runs. Allocate as `Box<[u8]>` and `heap::alloc` it so the - // alloc/free pair through the SAME Rust global allocator (mimalloc). - // Do NOT keep a `Vec`/`Box` binding past the FFI call — it would drop - // at scope exit and leave the queued microtask with a dangling pointer - // (UAF on read in `handle_data`, then double-free on drop). + // microtask runs. Allocate as `Box<[u8]>` so the alloc/free pair goes + // through the SAME Rust global allocator (mimalloc). + // + // Keep it as an owned `Box` (NOT leaked yet) until a success arm is + // reached. A `handshake`/`upgrade` handler can synchronously close the + // socket (see the note above), which makes the `!tcp.is_closed() && + // has_ws` re-check below fail and route into an else-arm that never + // calls `did_connect`; leaking the buffer up-front would then orphan it + // (no consumer to reclaim it). By only `heap::into_raw`-ing it at the + // moment of handoff, the else-arms drop the `Box` normally instead. let overflow_len = remain_buf.len(); - let overflow_ptr: *mut u8 = if overflow_len > 0 { + let mut overflow_box: Option> = if overflow_len > 0 { let mut v: Vec = Vec::new(); if v.try_reserve_exact(overflow_len).is_err() { // OOM here terminates with `invalid_response` rather than @@ -1581,11 +1638,18 @@ impl HTTPClient { return; } v.extend_from_slice(remain_buf); - // Leak across the FFI boundary; `InitialDataHandler` reconstructs - // the `Box<[u8]>` and drops it after delivery. - bun_core::heap::into_raw(v.into_boxed_slice()).cast::() + Some(v.into_boxed_slice()) } else { - core::ptr::null_mut() + None + }; + // Leak the owned buffer across the FFI boundary right before handoff; + // `InitialDataHandler` reconstructs the `Box<[u8]>` and drops it after + // delivery. Returns a null thin pointer when there is no overflow. + let take_overflow_ptr = |b: &mut Option>| -> *mut u8 { + match b.take() { + Some(boxed) => bun_core::heap::into_raw(boxed).cast::(), + None => core::ptr::null_mut(), + } }; // Check if we're using a proxy tunnel (wss:// through HTTP proxy) @@ -1613,7 +1677,10 @@ impl HTTPClient { // SAFETY: short-lived `&mut` for the field take. let ws = unsafe { (*this).outgoing_websocket.take().unwrap() }; - // Create the WebSocket client with the tunnel + // Create the WebSocket client with the tunnel. Hand off the + // overflow buffer (leak it across FFI) only now that we're + // committed to delivering it. + let overflow_ptr = take_overflow_ptr(&mut overflow_box); // SAFETY: live C++ back-reference. unsafe { (*ws).did_connect_with_tunnel( @@ -1672,6 +1739,10 @@ impl HTTPClient { // SAFETY: short-lived `&mut` for the field detach; ends before the FFI call below. unsafe { (*this).tcp.detach() }; if let uws::InternalSocket::Connected(native_socket) = socket.socket { + // Hand off the overflow buffer (leak it across FFI) only now + // that `did_connect` will consume it. If the socket is not + // Connected (else-arm), `overflow_box` is dropped instead. + let overflow_ptr = take_overflow_ptr(&mut overflow_box); // SAFETY: live C++ back-reference. unsafe { (*ws).did_connect( diff --git a/src/js/builtins/BunBuiltinNames.h b/src/js/builtins/BunBuiltinNames.h index 496ca899a10..69415c1ecbb 100644 --- a/src/js/builtins/BunBuiltinNames.h +++ b/src/js/builtins/BunBuiltinNames.h @@ -156,6 +156,7 @@ using namespace JSC; macro(processBindingConstants) \ macro(props) \ macro(pull) \ + macro(rawHeaders) \ macro(read) \ macro(readable) \ macro(readableType) \ @@ -179,6 +180,8 @@ using namespace JSC; macro(started) \ macro(state) \ macro(status) \ + macro(statusCode) \ + macro(statusMessage) \ macro(statusText) \ macro(stream) \ macro(structuredCloneForStream) \ diff --git a/src/js/thirdparty/ws.js b/src/js/thirdparty/ws.js index 25d68a95208..fe705bd2ba0 100644 --- a/src/js/thirdparty/ws.js +++ b/src/js/thirdparty/ws.js @@ -91,6 +91,42 @@ function emitWarning(type, message) { console.warn("[bun] Warning:", message); } +// ws emits `upgrade` / `unexpected-response` with an `http.IncomingMessage` for +// the handshake response. We bypass node:http, so build a minimal +// IncomingMessage-shaped Readable from the status + rawHeaders the native +// WebSocket parsed. Lazily pull in node:stream the first time it's needed. +let lazyReadable; +function makeHandshakeResponse(statusCode, statusMessage, rawHeaders, body) { + lazyReadable ??= require("node:stream").Readable; + const res = new lazyReadable({ read() {} }); + // Match node's http.IncomingMessage.headers: a plain object that inherits + // from Object.prototype (so `res.headers.hasOwnProperty(...)` works). Use + // Object.hasOwn for the dup-check so a header literally named "constructor" + // is not confused with Object.prototype.constructor. + const headers = (res.headers = {}); + res.rawHeaders = rawHeaders; + for (let i = 0; i < rawHeaders.length; i += 2) { + const lower = rawHeaders[i].toLowerCase(); + const value = rawHeaders[i + 1]; + const seen = Object.hasOwn(headers, lower); + if (lower === "set-cookie") { + if (!seen) headers[lower] = [value]; + else headers[lower].push(value); + } else { + headers[lower] = !seen ? value : headers[lower] + ", " + value; + } + } + res.statusCode = statusCode; + res.statusMessage = statusMessage; + res.httpVersion = "1.1"; + res.httpVersionMajor = 1; + res.httpVersionMinor = 1; + res.socket = res.connection = null; + if (body && body.length) res.push(body); + res.push(null); + return res; +} + // TODO: add private method on WebSocket to avoid these allocations function normalizeData(data, opts) { const isBinary = opts?.binary; @@ -125,6 +161,7 @@ class BunWebSocket extends EventEmitter { #paused = false; #fragments = false; #binaryType = "nodebuffer"; + #handshakeListenerRegistered = false; // Bitset to track whether event handlers are set. #eventId = 0; @@ -268,84 +305,91 @@ class BunWebSocket extends EventEmitter { } let ws = (this.#ws = new WebSocket(url, wsOptions)); ws.binaryType = "nodebuffer"; + // The native 'handshake' listener is registered lazily (from + // #armNativeBridge when the user subscribes to 'upgrade') so callers that + // only listen to 'open'/'message'/'close' never exercise the native + // handshake-dispatch path. return ws; } - #onOrOnce(event, listener, once) { - if (event === "unexpected-response" || event === "upgrade" || event === "redirect") { + #ensureHandshakeListener() { + if (this.#handshakeListenerRegistered) return; + this.#handshakeListenerRegistered = true; + this.#ws.addEventListener("handshake", event => this.#onHandshake(event.data), onceObject); + } + + #onHandshake(data) { + const { statusCode, statusMessage, rawHeaders } = data; + // The native client only forwards the successful 101 handshake here; it + // fails the connection on any other status before reaching this point. On a + // 101, bytes after the header block are the first WebSocket frame (not an + // HTTP body) and the native client forwards them to the protocol reader on + // connect, so don't include a body in the IncomingMessage. + const res = makeHandshakeResponse(statusCode, statusMessage, rawHeaders, null); + // ws emits `upgrade` with `(response)`, right before `open`. + this.emit("upgrade", res); + } + + // Arm one persistent native forwarder per ws event; EventEmitter fans the + // single emit out to every on/once listener. A `{once:true}` forwarder would + // auto-remove and let a later on() add a second, double-firing every emit. + #armNativeBridge(event) { + if (event === "unexpected-response" || event === "redirect") { emitWarning(event, "ws.WebSocket '" + event + "' event is not implemented in bun"); + return; } - const mask = 1 << eventIds[event]; - const hasPersistentListener = mask && (this.#eventId & mask) === mask; - // Add a native listener if: - // 1. For `on()`: no native listener exists yet (will be persistent) - // 2. For `once()`: no persistent `on()` listener exists (otherwise the persistent one forwards events) - // If only `once()` listeners exist, each needs its own native listener since they auto-remove - if (mask && !hasPersistentListener) { - // Only set the eventId bit for persistent `on` listeners, not for `once` - if (!once) { - this.#eventId |= mask; - } - if (event === "open") { - this.#ws.addEventListener( - "open", - () => { - this.emit("open"); - }, - once, - ); - } else if (event === "close") { - this.#ws.addEventListener( - "close", - ({ code, reason, wasClean }) => { - this.emit("close", code, reason, wasClean); - }, - once, - ); - } else if (event === "message") { - this.#ws.addEventListener( - "message", - ({ data }) => { - const isBinary = typeof data !== "string"; - if (isBinary) { - this.emit("message", this.#fragments ? [data] : data, isBinary); - } else { - let encoded = encoder.encode(data); - if (this.#binaryType !== "arraybuffer") { - encoded = Buffer.from(encoded.buffer, encoded.byteOffset, encoded.byteLength); - } - this.emit("message", this.#fragments ? [encoded] : encoded, isBinary); - } - }, - once, - ); - } else if (event === "error") { - this.#ws.addEventListener( - "error", - err => { - this.emit("error", err); - }, - once, - ); - } else if (event === "ping") { - this.#ws.addEventListener( - "ping", - ({ data }) => { - this.emit("ping", data); - }, - once, - ); - } else if (event === "pong") { - this.#ws.addEventListener( - "pong", - ({ data }) => { - this.emit("pong", data); - }, - once, - ); - } + if (event === "upgrade") { + // Lazily wire the native handshake listener; `upgrade` is emitted from + // #onHandshake. + this.#ensureHandshakeListener(); + return; } + const id = eventIds[event]; + // Not a native ws event (e.g. a user-defined one); EventEmitter handles it. + if (id === undefined) return; + const mask = 1 << id; + // Forwarder already armed for this event; the single emit reaches every listener. + if ((this.#eventId & mask) === mask) return; + this.#eventId |= mask; + if (event === "open") { + this.#ws.addEventListener("open", () => { + this.emit("open"); + }); + } else if (event === "close") { + this.#ws.addEventListener("close", ({ code, reason, wasClean }) => { + this.emit("close", code, reason, wasClean); + }); + } else if (event === "message") { + this.#ws.addEventListener("message", ({ data }) => { + const isBinary = typeof data !== "string"; + if (isBinary) { + this.emit("message", this.#fragments ? [data] : data, isBinary); + } else { + let encoded = encoder.encode(data); + if (this.#binaryType !== "arraybuffer") { + encoded = Buffer.from(encoded.buffer, encoded.byteOffset, encoded.byteLength); + } + this.emit("message", this.#fragments ? [encoded] : encoded, isBinary); + } + }); + } else if (event === "error") { + this.#ws.addEventListener("error", err => { + this.emit("error", err); + }); + } else if (event === "ping") { + this.#ws.addEventListener("ping", ({ data }) => { + this.emit("ping", data); + }); + } else if (event === "pong") { + this.#ws.addEventListener("pong", ({ data }) => { + this.emit("pong", data); + }); + } + } + + #onOrOnce(event, listener, once) { + this.#armNativeBridge(event); return once ? super.once(event, listener) : super.on(event, listener); } @@ -357,6 +401,26 @@ class BunWebSocket extends EventEmitter { return this.#onOrOnce(event, listener, onceObject); } + // `addListener` is an alias of `on`; `prependListener`/`prependOnceListener` + // add to the front of the listener list. ws / EventEmitter consumers reach + // for all of these, so each must arm the native bridge too — otherwise the + // handler sits on the EventEmitter list but the native event that drives + // `this.emit(...)` is never wired up and the callback silently never fires + // (e.g. `ws.addListener("upgrade", cb)`). + addListener(event, listener) { + return this.#onOrOnce(event, listener, undefined); + } + + prependListener(event, listener) { + this.#armNativeBridge(event); + return super.prependListener(event, listener); + } + + prependOnceListener(event, listener) { + this.#armNativeBridge(event); + return super.prependOnceListener(event, listener); + } + send(data, opts, cb) { if ($isCallable(opts)) { cb = opts; diff --git a/src/jsc/bindings/webcore/EventNames.h b/src/jsc/bindings/webcore/EventNames.h index dfd9feccd6c..293b389350a 100644 --- a/src/jsc/bindings/webcore/EventNames.h +++ b/src/jsc/bindings/webcore/EventNames.h @@ -29,16 +29,17 @@ namespace WebCore { -#define DOM_EVENT_NAMES_FOR_EACH(macro) \ - macro(error) \ - macro(abort) \ - macro(close) \ - macro(open) \ - macro(rename) \ - macro(message) \ - macro(change) \ - macro(messageerror) \ - macro(resourcetimingbufferfull) +#define DOM_EVENT_NAMES_FOR_EACH(macro) \ + macro(error) \ + macro(abort) \ + macro(close) \ + macro(open) \ + macro(rename) \ + macro(message) \ + macro(change) \ + macro(messageerror) \ + macro(handshake) \ + macro(resourcetimingbufferfull) struct EventNames { WTF_MAKE_NONCOPYABLE(EventNames); diff --git a/src/jsc/bindings/webcore/WebSocket.cpp b/src/jsc/bindings/webcore/WebSocket.cpp index 51f49057334..ec757e51b1a 100644 --- a/src/jsc/bindings/webcore/WebSocket.cpp +++ b/src/jsc/bindings/webcore/WebSocket.cpp @@ -74,6 +74,7 @@ #include #include "JSBuffer.h" +#include "BunClientData.h" #include "ErrorEvent.h" #include "WebSocketDeflate.h" @@ -1529,6 +1530,62 @@ void WebSocket::didReceiveBinaryData(const AtomString& eventName, const std::spa // }); } +void WebSocket::didReceiveHandshakeResponse(uint16_t statusCode, std::span statusMessage, std::span headers, std::span body) +{ + // The only consumer is the `ws` package shim, which registers a `handshake` + // listener on the native WebSocket before connecting when the user subscribes + // to the `upgrade` event. The browser-style `new WebSocket()` path never + // registers it, so skip the work entirely. + if (!this->hasEventListeners(eventNames().handshakeEvent)) + return; + + auto* context = scriptExecutionContext(); + if (!context) + return; + auto* globalObject = context->jsGlobalObject(); + auto& vm = globalObject->vm(); + auto scope = DECLARE_TOP_EXCEPTION_SCOPE(vm); + auto& builtinNames = WebCore::builtinNames(vm); + + auto* obj = JSC::constructEmptyObject(globalObject); + obj->putDirect(vm, builtinNames.statusCodePublicName(), JSC::jsNumber(statusCode)); + obj->putDirect(vm, builtinNames.statusMessagePublicName(), + JSC::jsString(vm, WTF::String({ statusMessage.data(), statusMessage.size() }))); + + // PicoHTTP already parsed the response head, so surface its header list as + // a flat [name, value, name, value, ...] rawHeaders array for the shim to + // fold into an http.IncomingMessage-shaped object. + auto* rawHeaders = JSC::constructEmptyArray(globalObject, nullptr, headers.size() * 2); + if (!rawHeaders || scope.exception()) [[unlikely]] { + scope.clearExceptionExceptTermination(); + return; + } + for (size_t i = 0; i < headers.size(); i++) { + rawHeaders->putDirectIndex(globalObject, i * 2, + JSC::jsString(vm, WTF::String({ headers[i].name_ptr, headers[i].name_len }))); + rawHeaders->putDirectIndex(globalObject, i * 2 + 1, + JSC::jsString(vm, WTF::String({ headers[i].value_ptr, headers[i].value_len }))); + } + obj->putDirect(vm, builtinNames.rawHeadersPublicName(), rawHeaders); + + JSC::JSUint8Array* bodyBuffer = createBuffer(globalObject, body); + if (!bodyBuffer || scope.exception()) [[unlikely]] { + scope.clearExceptionExceptTermination(); + return; + } + obj->putDirect(vm, builtinNames.bodyPublicName(), bodyBuffer); + + JSC::EnsureStillAliveScope ensureStillAlive(obj); + + MessageEvent::Init init; + init.data = obj; + init.origin = m_url.string(); + + this->incPendingActivityCount(); + dispatchEvent(MessageEvent::create(eventNames().handshakeEvent, WTF::move(init), EventIsTrusted::Yes)); + this->decPendingActivityCount(); +} + void WebSocket::didReceiveClose(CleanStatus wasClean, unsigned short code, WTF::String reason, bool isConnectionError) { // LOG(Network, "WebSocket %p didReceiveErrorMessage()", this); @@ -1900,6 +1957,11 @@ extern "C" void WebSocket__didConnectWithTunnel(WebCore::WebSocket* webSocket, v webSocket->didConnectWithTunnel(tunnel, bufferedData, len, deflate_params); } +extern "C" void WebSocket__didReceiveHandshakeResponse(WebCore::WebSocket* webSocket, uint16_t statusCode, const uint8_t* statusMessage, size_t statusMessageLen, const WebCore::WebSocket::HandshakeRawHeader* headers, size_t headersLen, const uint8_t* body, size_t bodyLen) +{ + webSocket->didReceiveHandshakeResponse(statusCode, std::span(statusMessage, statusMessageLen), std::span(headers, headersLen), std::span(body, bodyLen)); +} + extern "C" void WebSocket__didAbruptClose(WebCore::WebSocket* webSocket, Bun::WebSocketErrorCode errorCode) { webSocket->didFailWithErrorCode(errorCode); diff --git a/src/jsc/bindings/webcore/WebSocket.h b/src/jsc/bindings/webcore/WebSocket.h index a0c1ada560d..a241bb41543 100644 --- a/src/jsc/bindings/webcore/WebSocket.h +++ b/src/jsc/bindings/webcore/WebSocket.h @@ -130,6 +130,16 @@ class WebSocket final : public RefCounted, public EventTargetWithInli CLOSED = 3, }; + // A single handshake response header, borrowed from the buffer PicoHTTP + // parsed. Only valid for the duration of the didReceiveHandshakeResponse + // call that receives it. + struct HandshakeRawHeader { + const uint8_t* name_ptr; + size_t name_len; + const uint8_t* value_ptr; + size_t value_len; + }; + enum Opcode : unsigned char { Continue = 0x0, Text = 0x1, @@ -207,6 +217,7 @@ class WebSocket final : public RefCounted, public EventTargetWithInli void didReceiveMessage(String&& message); void didReceiveData(const char* data, size_t length); void didReceiveBinaryData(const AtomString& eventName, const std::span binaryData); + void didReceiveHandshakeResponse(uint16_t statusCode, std::span statusMessage, std::span headers, std::span body); void updateHasPendingActivity(); bool hasPendingActivity() const diff --git a/test/js/first_party/ws/ws-upgrade.test.ts b/test/js/first_party/ws/ws-upgrade.test.ts new file mode 100644 index 00000000000..bc7eb8a5f54 --- /dev/null +++ b/test/js/first_party/ws/ws-upgrade.test.ts @@ -0,0 +1,241 @@ +import { describe, expect, it } from "bun:test"; +import { once } from "events"; +import type { IncomingMessage } from "http"; +import { WebSocket, WebSocketServer } from "ws"; + +// https://github.com/oven-sh/bun/issues/31406 +// +// The `ws` client used to hardcode the `upgrade` event as "not implemented" +// and print a warning instead of firing it. Node's `ws` emits `upgrade` with +// the handshake response (an http.IncomingMessage) right before `open`. These +// tests use in-process servers (no subprocess spawning) so they run fast under +// the ASAN debug build. +describe("ws client upgrade event", () => { + it("fires before open with the 101 handshake response", async () => { + const wss = new WebSocketServer({ port: 0 }); + wss.on("connection", ws => ws.close()); + + const order: string[] = []; + const { promise, resolve, reject } = Promise.withResolvers(); + + const ws = new WebSocket("ws://localhost:" + wss.address().port); + try { + ws.on("upgrade", res => { + order.push("upgrade"); + resolve(res); + }); + ws.on("open", () => { + order.push("open"); + ws.close(); + }); + ws.on("error", reject); + + const res = await promise; + + // `upgrade` must fire, and it must fire before `open` (as in node's ws). + expect(order[0]).toBe("upgrade"); + // The argument is the handshake response (an http.IncomingMessage). + expect(res.statusCode).toBe(101); + expect(res.statusMessage).toBe("Switching Protocols"); + expect(res.httpVersion).toBe("1.1"); + expect(res.headers.upgrade?.toLowerCase()).toBe("websocket"); + expect(res.headers.connection?.toLowerCase()).toBe("upgrade"); + expect(typeof res.headers["sec-websocket-accept"]).toBe("string"); + expect(Array.isArray(res.rawHeaders)).toBe(true); + + await once(ws, "close"); + } finally { + ws.close(); + wss.close(); + } + }); + + it("exposes custom handshake response headers", async () => { + using server = Bun.serve({ + port: 0, + fetch(req, server) { + if ( + server.upgrade(req, { + headers: { "X-Custom-Header": "custom-value", "Set-Cookie": "a=1" }, + }) + ) { + return; + } + return new Response("no upgrade"); + }, + websocket: { open() {}, message() {} }, + }); + + const { promise, resolve, reject } = Promise.withResolvers(); + const ws = new WebSocket(server.url.href); + try { + ws.on("upgrade", resolve); + ws.on("error", reject); + // Close only after the connection is established to avoid racing the + // in-flight handshake (upgrade fires while still CONNECTING). + ws.on("open", () => ws.close()); + + const res = await promise; + expect(res.statusCode).toBe(101); + expect(res.headers["x-custom-header"]).toBe("custom-value"); + // node's IncomingMessage represents set-cookie as an array. + expect(res.headers["set-cookie"]).toEqual(["a=1"]); + + await once(ws, "close"); + } finally { + ws.close(); + } + }); + + it("supports once() for upgrade", async () => { + const wss = new WebSocketServer({ port: 0 }); + wss.on("connection", () => {}); + + const { promise, resolve, reject } = Promise.withResolvers(); + const ws = new WebSocket("ws://localhost:" + wss.address().port); + try { + ws.once("upgrade", resolve); + ws.on("open", () => ws.close()); + ws.on("error", reject); + const res = await promise; + expect(res.statusCode).toBe(101); + await once(ws, "close"); + } finally { + ws.close(); + wss.close(); + } + }); + + // ws / EventEmitter consumers also subscribe via addListener / + // prependListener / prependOnceListener; each must wire the native handshake + // listener so `upgrade` actually fires. + for (const method of ["addListener", "prependListener", "prependOnceListener"] as const) { + it(`fires upgrade when subscribed via ${method}`, async () => { + const wss = new WebSocketServer({ port: 0 }); + wss.on("connection", () => {}); + + const { promise, resolve, reject } = Promise.withResolvers(); + const ws = new WebSocket("ws://localhost:" + wss.address().port); + try { + ws[method]("upgrade", resolve); + ws.on("open", () => ws.close()); + ws.on("error", reject); + const res = await promise; + expect(res.statusCode).toBe(101); + await once(ws, "close"); + } finally { + ws.close(); + wss.close(); + } + }); + } + + // Node + ws emit `upgrade` and `open` from the same socket-data turn with no + // microtask checkpoint between them, so a microtask/nextTick queued inside the + // `upgrade` handler runs after the socket is OPEN. + it("does not drain microtasks between upgrade and open", async () => { + const wss = new WebSocketServer({ port: 0 }); + wss.on("connection", () => {}); + + const { promise, resolve, reject } = Promise.withResolvers<{ microtask: number; nextTick: number }>(); + const ws = new WebSocket("ws://localhost:" + wss.address().port); + try { + const states: { microtask?: number; nextTick?: number } = {}; + ws.on("upgrade", () => { + // These are scheduled while still CONNECTING but must observe OPEN, + // because `open` fires before the microtask/nextTick checkpoint. + queueMicrotask(() => { + states.microtask = ws.readyState; + }); + process.nextTick(() => { + states.nextTick = ws.readyState; + }); + }); + ws.on("open", async () => { + // After a macrotask turn both the microtask and the nextTick have run. + await Bun.sleep(0); + resolve(states as { microtask: number; nextTick: number }); + }); + ws.on("error", reject); + + const seen = await promise; + expect(seen.microtask).toBe(WebSocket.OPEN); + expect(seen.nextTick).toBe(WebSocket.OPEN); + + ws.close(); + await once(ws, "close"); + } finally { + ws.close(); + wss.close(); + } + }); + + // Consequence of the ordering above: a `process.nextTick(() => ws.send(...))` + // scheduled from the `upgrade` handler runs after `open`, so the socket is + // OPEN and the send is delivered (rather than hitting InvalidStateError while + // still CONNECTING). + it("delivers a send scheduled from the upgrade handler", async () => { + const wss = new WebSocketServer({ port: 0 }); + const { promise, resolve, reject } = Promise.withResolvers(); + wss.on("connection", server => { + server.on("message", data => resolve(data.toString())); + }); + + const ws = new WebSocket("ws://localhost:" + wss.address().port); + try { + ws.on("upgrade", () => { + process.nextTick(() => ws.send("from-upgrade")); + }); + ws.on("error", reject); + + const received = await promise; + expect(received).toBe("from-upgrade"); + + ws.close(); + await once(ws, "close"); + } finally { + ws.close(); + wss.close(); + } + }); +}); + +// The native forwarder is armed once per event type; EventEmitter fans each +// emit out to every on/once listener. A once-style registration followed by a +// persistent on() must not install a second forwarder (that would emit twice). +describe("ws client event listener bridging", () => { + for (const register of ["once", "prependOnceListener"] as const) { + it(`${register}() + on() for the same event each fire exactly once`, async () => { + const wss = new WebSocketServer({ port: 0 }); + wss.on("connection", () => {}); + + const ws = new WebSocket("ws://localhost:" + wss.address().port); + try { + const { promise, resolve, reject } = Promise.withResolvers(); + let onceCount = 0; + let onCount = 0; + ws[register]("open", () => { + onceCount++; + }); + ws.on("open", () => { + onCount++; + resolve(); + }); + ws.on("error", reject); + + await promise; + // A duplicate native forwarder re-emits within the same dispatch, so + // onCount is already final here; a macrotask turn makes that certain. + await Bun.sleep(0); + expect(onceCount).toBe(1); + expect(onCount).toBe(1); + + ws.close(); + await once(ws, "close"); + } finally { + ws.close(); + wss.close(); + } + }); + } +});