feature: Harden WebSocket handshake (version check) + Node backpressure

- Add Sec-WebSocket-Version validation via a shared WebSocketHandshake.validate:
  missing key -> 400; version absent -> 400; version present but != 13 -> 426
  (advertising Sec-WebSocket-Version: 13). Used by both manual backends (Native,
  Node), dedup'ing the key check too.
- Node: handle write backpressure — pause the socket when socket.write() returns
  false, resume on 'drain' — so a slow consumer can't make us buffer unbounded.
- Node writeHttpClose now copies the response's own headers (so the 426's
  Sec-WebSocket-Version reaches the client; Native's serialize already did).
- Tests: Native + Node "unsupported version -> 426". JVM 1635 / Native 1388 /
  JS 1387 pass.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: xerial

plans/2026-06-19-websocket-hardening.md

@@ -0,0 +1,26 @@
+# WebSocket hardening (Sec-WebSocket-Version + Node backpressure)
+
+## Context
+Follow-up hardening for the WebSocket servers (Native #576, Node #577). Two gaps noted at merge:
+1. Neither manual backend validated `Sec-WebSocket-Version` (RFC 6455 says reply `426` advertising the
+   supported version on a mismatch).
+2. Node ignored `socket.write()` backpressure (could buffer unbounded for a slow consumer).
+
+## Changes
+- **`WebSocketHandshake` (shared)** — `validate(request): Either[Response, String]`: missing/empty
+  `Sec-WebSocket-Key` → 400; `Sec-WebSocket-Version` absent → 400, present-but-not-13 → 426 (with
+  `Sec-WebSocket-Version: 13`); otherwise `Right(key)`. Dedups the key check and adds the version
+  check across both manual backends. Native (`NativeHttpServer.handleWebSocketUpgrade`) and Node
+  (`NodeWebSocket.handleUpgrade`) now call it.
+- **Node backpressure** — `NodeWebSocketContext.writeFrame` pauses the socket when `write()` returns
+  false; `accept` wires `'drain'` → `socket.resume()`. Bounds memory under a slow consumer.
+- **Node `writeHttpClose`** now copies the response's own headers (so the 426's `Sec-WebSocket-Version`
+  reaches the client; Native's `serialize` already did).
+
+## Verification
+`scalafmt`; full JVM/JS/Native suites + Netty compile. New tests: Native + Node "unsupported version
+→ 426 (advertising 13)". JVM 1635 / Native 1388 / JS 1387 pass.
+
+## Follow-ups (remaining)
+Portable Native idle/read timeout; Native libcurl client bug; cross-platform WebSocket client;
+permessage-deflate.

uni/.js/src/main/scala/wvlet/uni/http/NodeWebSocket.scala

@@ -47,11 +47,11 @@ private[http] object NodeWebSocket extends LogSupport:
         case None =>
           destroyQuietly(socket)
         case Some(route) =>
-          request.header(HttpHeader.SecWebSocketKey).filter(_.nonEmpty) match
-            case None =>
-              // A WebSocket upgrade without a Sec-WebSocket-Key is malformed (RFC 6455 §4.2.1).
-              writeHttpClose(socket, Response.badRequest("Missing Sec-WebSocket-Key"))
-            case Some(key) =>
+          WebSocketHandshake.validate(request) match
+            case Left(rejection) =>
+              // Malformed handshake: missing key (400), or unsupported Sec-WebSocket-Version (426).
+              writeHttpClose(socket, rejection)
+            case Right(key) =>
               gateAndAccept(socket, head, key, route, request, maxFrameSize)
     catch
       case NonFatal(e) =>
@@ -147,8 +147,17 @@ private[http] object NodeWebSocket extends LogSupport:
       val onData: js.Function1[js.Dynamic, Unit] =
         (chunk: js.Dynamic) => drive(NodeBytes.toBytes(chunk))
       val onClose: js.Function1[js.Dynamic, Unit] = (_: js.Dynamic) => notifyClose()
+      // Resume reading once a back-pressured write has drained (see NodeWebSocketContext.writeFrame).
+      val onDrain: js.Function0[Unit] =
+        () =>
+          try
+            socket.applyDynamic("resume")()
+          catch
+            case NonFatal(_) =>
+              ()
       socket.applyDynamic("on")("close", onClose)
       socket.applyDynamic("on")("error", onClose)
+      socket.applyDynamic("on")("drain", onDrain)
 
       try
         handler.onOpen(ctx)
@@ -213,6 +222,17 @@ private[http] object NodeWebSocket extends LogSupport:
         .append(" ")
         .append(response.status.reason)
         .append("\r\n")
+      // Copy the response's own headers (e.g. Sec-WebSocket-Version on a 426), minus the ones set
+      // explicitly below.
+      response
+        .headers
+        .entries
+        .foreach { case (name, value) =>
+          if !name.equalsIgnoreCase(HttpHeader.Connection) &&
+            !name.equalsIgnoreCase(HttpHeader.ContentLength)
+          then
+            sb.append(name).append(": ").append(value).append("\r\n")
+        }
       sb.append(HttpHeader.Connection).append(": close\r\n")
       sb.append(HttpHeader.ContentLength).append(": ").append(body.length).append("\r\n\r\n")
       val head = sb.toString.getBytes(StandardCharsets.ISO_8859_1)
@@ -279,9 +299,14 @@ private[http] class NodeWebSocketContext(socket: js.Dynamic, override val reques
 
   private def writeFrame(opcode: Int, payload: Array[Byte]): Unit =
     try
-      socket.applyDynamic("write")(
-        NodeBytes.toUint8Array(WebSocketFrame.encodeFrame(opcode, payload))
-      )
+      val accepted =
+        socket.applyDynamic("write")(
+          NodeBytes.toUint8Array(WebSocketFrame.encodeFrame(opcode, payload))
+        )
+      // Backpressure: write() returns false when the kernel send buffer is full. Pause reading so a
+      // slow consumer can't make us buffer unbounded; the 'drain' listener (wired in accept) resumes.
+      if !accepted.asInstanceOf[Boolean] then
+        socket.applyDynamic("pause")()
     catch
       case NonFatal(e) =>
         debug(s"WebSocket write failed: ${e.getMessage}")

uni/.js/src/test/scala/wvlet/uni/http/NodeWebSocketTest.scala

@@ -63,13 +63,13 @@ class NodeWebSocketTest extends UniTest:
     socket.applyDynamic("on")("error", onError)
 
     /** Send the upgrade request; resolves with the parsed handshake response. */
-    def connect(path: String): Future[Handshake] =
+    def connect(path: String, version: String = "13"): Future[Handshake] =
       val onConnect: js.Function0[Unit] =
         () =>
           val req =
             s"GET ${path} HTTP/1.1\r\nHost: localhost\r\n" +
               s"Upgrade: websocket\r\nConnection: Upgrade\r\n" +
-              s"Sec-WebSocket-Key: ${key}\r\nSec-WebSocket-Version: 13\r\n\r\n"
+              s"Sec-WebSocket-Key: ${key}\r\nSec-WebSocket-Version: ${version}\r\n\r\n"
           socket.applyDynamic("write")(
             NodeBytes.toUint8Array(req.getBytes(StandardCharsets.ISO_8859_1))
           )
@@ -313,6 +313,23 @@ class NodeWebSocketTest extends UniTest:
       }
   }
 
+  test("WebSocket upgrade with an unsupported version is rejected with 426") {
+    NodeServer
+      .withPort(0)
+      .withWebSocketRoute("/ws") { _ =>
+        new WebSocketHandler {}
+      }
+      .startAndAwait { server =>
+        val client = NodeWsClient(server.localPort)
+        Rx.future(client.connect("/ws", version = "8"))
+          .map { hs =>
+            client.close()
+            hs.status shouldBe 426
+            hs.headers.get("sec-websocket-version") shouldBe Some("13")
+          }
+      }
+  }
+
   test("WebSocket upgrade can be rejected by a filter") {
     val deny = RxHttpFilter { (_, _) =>
       Rx.single(Response.forbidden)

uni/.native/src/main/scala/wvlet/uni/http/NativeHttpServer.scala

@@ -263,18 +263,14 @@ class NativeHttpServer(config: NativeServerConfig) extends HttpServer with LogSu
     * worker for the WebSocket's lifetime).
     */
   private def handleWebSocketUpgrade(clientFd: Int, request: Request, route: WebSocketRoute): Unit =
-    request.header(HttpHeader.SecWebSocketKey).filter(_.nonEmpty) match
-      case None =>
-        // A WebSocket upgrade without a Sec-WebSocket-Key is a malformed handshake (RFC 6455 §4.2.1).
+    WebSocketHandshake.validate(request) match
+      case Left(rejection) =>
+        // Malformed handshake: missing key (400), or unsupported Sec-WebSocket-Version (426).
         NativeSocket.sendAll(
           clientFd,
-          HttpResponseWriter.serialize(
-            Response.badRequest("Missing Sec-WebSocket-Key"),
-            keepAlive = false,
-            includeBody = true
-          )
+          HttpResponseWriter.serialize(rejection, keepAlive = false, includeBody = true)
         )
-      case Some(key) =>
+      case Right(key) =>
         // Capture the request as threaded through the filter, so attributes a filter adds during
         // the handshake reach the WebSocketContext (matching the Netty backend).
         val upgradeRequest = AtomicReference[Request](request)

uni/.native/src/test/scala/wvlet/uni/http/NativeServerTest.scala

@@ -443,6 +443,23 @@ class NativeServerTest extends UniTest:
       }
   }
 
+  test("WebSocket upgrade with an unsupported version is rejected with 426") {
+    NativeServer
+      .withPort(0)
+      .withWebSocketRoute("/ws") { _ =>
+        new WebSocketHandler {}
+      }
+      .start { server =>
+        val resp = request(
+          server.localPort,
+          "GET /ws HTTP/1.1\r\nHost: localhost\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n" +
+            "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\nSec-WebSocket-Version: 8\r\n\r\n"
+        )
+        resp.status shouldBe 426
+        resp.headers.get("sec-websocket-version") shouldBe Some("13")
+      }
+  }
+
   test("WebSocket upgrade can be rejected by a filter") {
     val deny = RxHttpFilter { (_, _) =>
       wvlet.uni.rx.Rx.single(Response.forbidden)

uni/src/main/scala/wvlet/uni/http/WebSocketHandshake.scala

@@ -0,0 +1,50 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package wvlet.uni.http
+
+/**
+  * Shared validation of a WebSocket upgrade request, used by the backends that perform the
+  * handshake by hand (Native, Node.js). Centralizes the `Sec-WebSocket-Key` and
+  * `Sec-WebSocket-Version` checks so both backends reject malformed handshakes identically.
+  */
+private[http] object WebSocketHandshake:
+
+  /** The only WebSocket protocol version this server speaks (RFC 6455). */
+  final val SupportedVersion = "13"
+
+  /**
+    * Validate the upgrade headers. `Right(key)` carries the `Sec-WebSocket-Key` when the handshake
+    * is acceptable; `Left(response)` is the rejection to send:
+    *   - 400 if `Sec-WebSocket-Key` is missing/empty or `Sec-WebSocket-Version` is absent,
+    *   - 426 (advertising `Sec-WebSocket-Version: 13`) if the requested version is not 13.
+    */
+  def validate(request: Request): Either[Response, String] =
+    request.header(HttpHeader.SecWebSocketKey).filter(_.nonEmpty) match
+      case None =>
+        Left(Response.badRequest("Missing Sec-WebSocket-Key"))
+      case Some(key) =>
+        request.header(HttpHeader.SecWebSocketVersion).map(_.trim) match
+          case Some(SupportedVersion) =>
+            Right(key)
+          case Some(_) =>
+            // Unsupported version: tell the client which version we speak (RFC 6455 §4.2.2 / 4.4).
+            Left(
+              Response(HttpStatus.UpgradeRequired_426)
+                .addHeader(HttpHeader.SecWebSocketVersion, SupportedVersion)
+                .withTextContent("Unsupported Sec-WebSocket-Version")
+            )
+          case None =>
+            Left(Response.badRequest("Missing Sec-WebSocket-Version"))
+
+end WebSocketHandshake