fix(http): give websocket upgrades to JS listeners

changelog.d/11084-http-websocket-upgrade-listener.md

@@ -0,0 +1 @@
+Fixed HTTP WebSocket upgrades claimed by JavaScript `upgrade` listeners. Perry now hands public packages such as `ws` the untouched `net.Socket` and a binary `Buffer` for the upgrade head, including when that head is empty, so `WebSocketServer` can complete its own handshake and emit `connection`.

crates/perry-ext-http/src/server/raw_upgrade.rs

@@ -10,22 +10,24 @@
 //! the listener's handwritten 101 would both reach the client, and the
 //! unconsumed body bytes (`head`) were lost.
 //!
-//! This module adds the Node-exact path for *keyless* Upgrade requests
-//! (no `Sec-WebSocket-Key` — i.e. not a real WebSocket client handshake):
+//! This module adds the Node-exact path for Upgrade requests claimed by an
+//! `'upgrade'` listener:
 //!
 //! 1. When the server has `'upgrade'` listeners, the accept task peeks the
 //!    request head off the TCP stream *before* handing anything to hyper.
-//! 2. If the head carries `Connection: …upgrade…` + an `Upgrade:` header and
-//!    no `Sec-WebSocket-Key`, the stream is handed to perry-ext-net
+//! 2. If the head carries `Connection: …upgrade…` + an `Upgrade:` header, the
+//!    stream is handed to perry-ext-net
 //!    (`adopt_upgraded_tcp_stream`) so JS sees a standard `net.Socket`
 //!    surface, and the `'upgrade'` listeners fire with the unconsumed bytes
 //!    after the head as `head`.
-//! 3. Anything else (no Upgrade header, real WS handshakes, oversized or
-//!    truncated heads) is replayed to hyper byte-for-byte through
-//!    `PrefixedStream`, preserving today's behavior.
+//! 3. Anything else (no Upgrade header, oversized or truncated heads) is
+//!    replayed to hyper byte-for-byte through `PrefixedStream`, preserving
+//!    today's behavior.
 //!
-//! Real WebSocket handshakes (key present) deliberately keep the
-//! tungstenite path so `new WebSocketServer({ server })` keeps working.
+//! Native attached WebSocket servers have no JS `'upgrade'` listener and keep
+//! the internal WebSocket path. A listener, including the one installed by
+//! the public `ws` package, owns the handshake and must receive the untouched
+//! socket even when `Sec-WebSocket-Key` is present.
 
 use std::collections::HashMap;
 use std::net::SocketAddr;
@@ -116,6 +118,18 @@ fn find_head_end(buf: &[u8]) -> Option<usize> {
     buf.windows(4).position(|w| w == b"\r\n\r\n").map(|p| p + 4)
 }
 
+fn is_upgrade_head(headers: &HashMap<String, String>) -> bool {
+    let connection_upgrade = headers
+        .get("connection")
+        .map(|value| {
+            value
+                .split(',')
+                .any(|token| token.trim().eq_ignore_ascii_case("upgrade"))
+        })
+        .unwrap_or(false);
+    connection_upgrade && headers.contains_key("upgrade")
+}
+
 /// Peek the request head and dispatch a raw `'upgrade'` if it qualifies.
 /// Only called when the server has `'upgrade'` listeners.
 pub(crate) async fn peek_and_maybe_dispatch_raw_upgrade(
@@ -147,13 +161,7 @@ pub(crate) async fn peek_and_maybe_dispatch_raw_upgrade(
         return PeekResult::Passthrough(PrefixedStream::new(buf, stream));
     };
 
-    let connection_upgrade = headers_lower
-        .get("connection")
-        .map(|v| v.to_ascii_lowercase().contains("upgrade"))
-        .unwrap_or(false);
-    let has_upgrade = headers_lower.contains_key("upgrade");
-    let has_ws_key = headers_lower.contains_key("sec-websocket-key");
-    if !connection_upgrade || !has_upgrade || has_ws_key {
+    if !is_upgrade_head(&headers_lower) {
         return PeekResult::Passthrough(PrefixedStream::new(buf, stream));
     }
 
@@ -229,3 +237,28 @@ fn parse_head(
     }
     Some((method, url, headers_lower, raw_headers))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{is_upgrade_head, parse_head};
+
+    fn parsed_headers(head: &[u8]) -> std::collections::HashMap<String, String> {
+        parse_head(head).expect("valid request head").2
+    }
+
+    #[test]
+    fn websocket_handshake_belongs_to_the_upgrade_listener() {
+        let headers = parsed_headers(
+            b"GET / HTTP/1.1\r\nHost: localhost\r\nConnection: keep-alive, Upgrade\r\nUpgrade: websocket\r\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n\r\n",
+        );
+
+        assert!(is_upgrade_head(&headers));
+    }
+
+    #[test]
+    fn ordinary_request_stays_on_the_http_path() {
+        let headers = parsed_headers(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n");
+
+        assert!(!is_upgrade_head(&headers));
+    }
+}

crates/perry-ext-http/src/server/server.rs

@@ -664,8 +664,8 @@ fn serve_http_connection(
     }
     tokio::spawn(async move {
         // #4973 — when `'upgrade'` listeners exist, peek the
-        // request head before hyper writes anything: a keyless
-        // Upgrade request must reach JS as a raw net.Socket
+        // request head before hyper writes anything: an Upgrade
+        // request claimed by JS must reach it as a raw net.Socket
         // with NO response on the wire (Node semantics). Other
         // connections replay the peeked bytes to hyper.
         let has_upgrade_listeners = get_handle::<HttpServer>(server_handle)
@@ -1278,9 +1278,9 @@ async fn handle_request(
     // `'request'` when the server has no `'upgrade'` listeners — the
     // unconditional branch used to hijack it into a bogus 101; (b) only a
     // real WebSocket handshake (`Sec-WebSocket-Key` present) belongs on the
-    // tungstenite path — keyless Upgrade requests are served Node-style by
-    // the raw peek path in raw_upgrade.rs and only reach hyper when no
-    // listener was attached at accept time.
+    // native path only for an attached native WebSocket server. JS `'upgrade'`
+    // listeners own their handshake and are served Node-style by the raw peek
+    // path in raw_upgrade.rs.
     if crate::server::upgrade::is_websocket_upgrade(&req) {
         let has_upgrade_listeners = get_handle::<HttpServer>(server_handle)
             .map(|server| server_has_event_listener(server, "upgrade"))

crates/perry-ext-http/src/server/upgrade.rs

@@ -17,20 +17,18 @@
 //!    standard `ws_id` that the rest of perry-ext-ws's surface
 //!    consumes.
 //! 4. The `'upgrade'` listeners on the HTTP server are fired with
-//!    `(im_f64, ws_id_f64, head_str_f64)`. `ws_id_f64` is the same
+//!    `(im_f64, ws_id_f64, head_buffer_f64)`. `ws_id_f64` is the same
 //!    integer id as standalone `WebSocketServer({port})` connections,
 //!    so user code can interact with it through `ws.on('message',…)`,
 //!    `ws.send(…)`, `ws.close(…)` unchanged.
 //!
 //! Attached WebSocket servers are native observers registered by perry-ext-ws.
 
-use perry_ffi::{alloc_string, get_handle_mut, JsClosure, RawClosureHeader};
+use perry_ffi::{get_handle_mut, JsClosure, RawClosureHeader};
 
 use crate::server::request::handle_to_pointer_f64;
 use crate::server::server::HttpServer;
-use crate::server::types::{
-    js_promise_run_microtasks, POINTER_TAG, PTR_MASK, STRING_TAG, TAG_UNDEFINED,
-};
+use crate::server::types::{js_promise_run_microtasks, POINTER_TAG, PTR_MASK};
 
 /// Test whether a request looks like a WebSocket upgrade — checks
 /// `Connection: Upgrade` (case-insensitive contains) and
@@ -51,6 +49,11 @@ pub(crate) fn is_websocket_upgrade(req: &hyper::Request<hyper::body::Incoming>)
     connection_ok && upgrade_ok
 }
 
+fn upgrade_head_arg(head_data: &[u8]) -> f64 {
+    let head = perry_ffi::alloc_buffer(head_data);
+    f64::from_bits(POINTER_TAG | (head as u64 & PTR_MASK))
+}
+
 /// Fire the `'upgrade'` event listeners with `(im, wsId, head)`.
 /// Called from the main-thread event loop after the upgrade pending
 /// has been dispatched.
@@ -79,14 +82,10 @@ pub(crate) fn fire_upgrade_listeners(
     // (1.0_f64) would have bits 0x3FF0_…, which `unbox_to_i64`
     // AND-masks to 0, missing the WS_CONNECTIONS lookup entirely.
     let ws_id_f64 = f64::from_bits(POINTER_TAG | (ws_id as u64 & PTR_MASK));
-    let head_str = if head_data.is_empty() {
-        f64::from_bits(TAG_UNDEFINED)
-    } else {
-        let s = String::from_utf8_lossy(&head_data).into_owned();
-        let header = alloc_string(&s);
-        f64::from_bits(STRING_TAG | (header.as_raw() as u64 & PTR_MASK))
-    };
-    let head_str = scope.root_nanbox(head_str);
+    // Node always supplies a Buffer, including for a zero-length head. Public
+    // `ws` reads `head.length` before deciding whether to call `unshift`, and
+    // upgrade bytes are arbitrary protocol data rather than UTF-8 text.
+    let head_arg = scope.root_nanbox(upgrade_head_arg(&head_data));
 
     for cb in listeners {
         if cb.get() == 0 {
@@ -96,7 +95,7 @@ pub(crate) fn fire_upgrade_listeners(
             let raw = cb.get() as *const RawClosureHeader;
             let closure = JsClosure::from_raw(raw);
             if !closure.is_null() {
-                let _ = closure.call3(req_f64, ws_id_f64, head_str.get());
+                let _ = closure.call3(req_f64, ws_id_f64, head_arg.get());
             }
             js_promise_run_microtasks();
         }
@@ -112,6 +111,28 @@ fn _force_link() -> u64 {
     POINTER_TAG | (PTR_MASK & 0)
 }
 
+#[cfg(test)]
+mod tests {
+    use super::{upgrade_head_arg, POINTER_TAG, PTR_MASK};
+
+    fn head_bytes(data: &[u8]) -> &'static [u8] {
+        let arg = upgrade_head_arg(data);
+        assert_eq!(arg.to_bits() & !PTR_MASK, POINTER_TAG);
+        let ptr = (arg.to_bits() & PTR_MASK) as *const perry_ffi::BufferHeader;
+        perry_ffi::read_buffer_bytes(ptr).expect("upgrade head buffer")
+    }
+
+    #[test]
+    fn empty_upgrade_head_is_an_empty_buffer() {
+        assert_eq!(head_bytes(&[]), &[] as &[u8]);
+    }
+
+    #[test]
+    fn upgrade_head_preserves_binary_bytes() {
+        assert_eq!(head_bytes(&[0xff, 0x00, 0x80]), &[0xff, 0x00, 0x80]);
+    }
+}
+
 /// Read owned address metadata without allocating JS objects or introducing a
 /// reverse dependency from ws to HTTP.
 pub(crate) fn attached_address(handle: i64) -> Option<(String, u16)> {