Repository: denoland/deno
Pull request: fix(ext/http): allow rejecting HTTP upgrades with non-101 status codes

Previously, when using node:http's upgrade event, attempting to reject an upgrade request with a non-101 status code (e.g., 401 Unauthorized) would throw a "write UNKNOWN" error because the UpgradeStream parser only accepted 101 Switching Protocols responses.

Now, non-101 responses are handled correctly:
- The rejection response is sent through hyper's normal response path
- The connection is properly closed after sending the response
- The server continues running and can handle subsequent requests

This matches Node.js behavior where servers can reject upgrade requests by writing any valid HTTP response to the socket.

Fixes #32311

Changed files:
filename: ext/http/http_next.rs
patch: @@ -1368,13 +1368,19 @@ enum UpgradeStreamWriteState {
     AsyncMut<Option<(NetworkStreamReadHalf, Bytes)>>,
   ),
   Network(NetworkStreamWriteHalf),
+  /// The upgrade was rejected with a non-101 status code.
+  /// The response has been sent and the stream is now closed for writing.
+  Rejected,
   Failed,
 }
 
 struct UpgradeStream {
   read: Rc<AsyncRefCell<Option<(NetworkStreamReadHalf, Bytes)>>>,
   write: AsyncRefCell<UpgradeStreamWriteState>,
   cancel_handle: CancelHandle,
+  /// Set to true when the upgrade was rejected with a non-101 status.
+  /// When rejected, reads return EOF and writes are silently ignored.
+  rejected: std::cell::Cell<bool>,
 }
 
 impl UpgradeStream {
@@ -1386,13 +1392,19 @@ impl UpgradeStream {
       read,
       write: AsyncRefCell::new(write),
       cancel_handle: CancelHandle::new(),
+      rejected: std::cell::Cell::new(false),
     }
   }
 
   async fn read(
     self: Rc<Self>,
     buf: &mut [u8],
   ) -> Result<usize, std::io::Error> {
+    // If the upgrade was rejected, return EOF
+    if self.rejected.get() {
+      return Ok(0);
+    }
+
     let cancel_handle = RcRef::map(self.clone(), |this| &this.cancel_handle);
     async {
       let read = RcRef::map(self, |this| &this.read);
@@ -1414,13 +1426,20 @@ impl UpgradeStream {
 
   async fn write(self: Rc<Self>, buf: &[u8]) -> Result<usize, std::io::Error> {
     let cancel_handle = RcRef::map(self.clone(), |this| &this.cancel_handle);
+    let this = self.clone();
     async {
       let wr = RcRef::map(self, |this| &this.write);
       let mut wr = wr.borrow_mut().await;
       match std::mem::replace(&mut *wr, UpgradeStreamWriteState::Failed) {
         UpgradeStreamWriteState::Failed => {
           Err(std::io::Error::other(HttpNextError::RawUpgradeFailed))
         }
+        UpgradeStreamWriteState::Rejected => {
+          // The upgrade was rejected and the response was already sent.
+          // Silently accept writes but don't do anything with them.
+          *wr = UpgradeStreamWriteState::Rejected;
+          Ok(buf.len())
+        }
         UpgradeStreamWriteState::Parsing(
           mut bytes,
           http,
@@ -1440,36 +1459,69 @@ impl UpgradeStream {
               Ok(buf.len())
             }
             Ok(httparse::Status::Complete(n)) => {
-              if response.code != Some(StatusCode::SWITCHING_PROTOCOLS.as_u16())
-              {
-                return Err(std::io::Error::other(
-                  HttpNextError::InvalidHttpStatusLine,
-                ));
-              }
-
-              http
-                .otel_info_set_status(StatusCode::SWITCHING_PROTOCOLS.as_u16());
-              http.response_parts().status = StatusCode::SWITCHING_PROTOCOLS;
+              let status_code = response.code.unwrap_or(200);
 
-              for header in response.headers {
-                http.response_parts().headers.append(
-                  HeaderName::from_bytes(header.name.as_bytes())
-                    .map_err(std::io::Error::other)?,
-                  HeaderValue::from_bytes(header.value)
-                    .map_err(std::io::Error::other)?,
+              if status_code == StatusCode::SWITCHING_PROTOCOLS.as_u16() {
+                // Upgrade accepted - proceed with upgrade
+                http.otel_info_set_status(
+                  StatusCode::SWITCHING_PROTOCOLS.as_u16(),
                 );
+                http.response_parts().status = StatusCode::SWITCHING_PROTOCOLS;
+
+                for header in response.headers {
+                  http.response_parts().headers.append(
+                    HeaderName::from_bytes(header.name.as_bytes())
+                      .map_err(std::io::Error::other)?,
+                    HeaderValue::from_bytes(header.value)
+                      .map_err(std::io::Error::other)?,
+                  );
+                }
+
+                http.complete();
+
+                let upgraded =
+                  on_upgrade.await.map_err(std::io::Error::other)?;
+                let (stream, bytes) = extract_network_stream(upgraded);
+                let (read, write) = stream.into_split();
+
+                let _ = read_cell.insert((read, bytes));
+                *wr = UpgradeStreamWriteState::Network(write);
+
+                Ok(n - prev_len)
+              } else {
+                // Upgrade rejected - send the rejection response through hyper
+                http.otel_info_set_status(status_code);
+                http.response_parts().status =
+                  StatusCode::from_u16(status_code)
+                    .unwrap_or(StatusCode::BAD_REQUEST);
+
+                for header in response.headers {
+                  http.response_parts().headers.append(
+                    HeaderName::from_bytes(header.name.as_bytes())
+                      .map_err(std::io::Error::other)?,
+                    HeaderValue::from_bytes(header.value)
+                      .map_err(std::io::Error::other)?,
+                  );
+                }
+
+                // Any data after the headers is the response body
+                let body = bytes.split_off(n);
+                if !body.is_empty() {
+                  http.set_response_body(ResponseBytesInner::Bytes(
+                    BufView::from(body.freeze()),
+                  ));
+                }
+
+                http.complete();
+
+                // Mark as rejected - no upgrade will happen
+                *wr = UpgradeStreamWriteState::Rejected;
+                this.rejected.set(true);
+                // Drop the on_upgrade future since we're not upgrading
+                drop(on_upgrade);
+
+                Ok(buf.len())
               }
-
-              http.complete();
-
-              let upgraded = on_upgrade.await.map_err(std::io::Error::other)?;
-              let (stream, bytes) = extract_network_stream(upgraded);
-              let (read, write) = stream.into_split();
-
-              let _ = read_cell.insert((read, bytes));
-              *wr = UpgradeStreamWriteState::Network(write);
-
-              Ok(n - prev_len)
             }
             Err(e) => Err(std::io::Error::other(e)),
           }
@@ -1496,7 +1548,12 @@ impl UpgradeStream {
       UpgradeStreamWriteState::Failed => {
         Err(std::io::Error::other(HttpNextError::RawUpgradeFailed))
       }
+      UpgradeStreamWriteState::Rejected => {
+        // The upgrade was rejected; silently accept writes
+        Ok(buf1.len() + buf2.len())
+      }
       UpgradeStreamWriteState::Parsing(..) => {
+        drop(wr);
         self.write(if buf1.is_empty() { buf2 } else { buf1 }).await
       }
       UpgradeStreamWriteState::Network(stream) => {

filename: tests/unit_node/http_test.ts
patch: @@ -2316,3 +2316,41 @@ Deno.test("[node/http] ServerResponse.writeEarlyHints", async () => {
   });
   await promise;
 });
+
+// https://github.com/denoland/deno/issues/32311
+Deno.test("[node/http] upgrade request can be rejected with non-101 status", async () => {
+  const { promise, resolve } = Promise.withResolvers<void>();
+  const server = http.createServer((_req, res) => res.end("ok"));
+
+  server.on("upgrade", (_req, socket) => {
+    const msg = "HTTP/1.1 401 Unauthorized\r\n" +
+      "Connection: close\r\n" +
+      "Content-Length: 0\r\n" +
+      "\r\n";
+    socket.end(msg);
+  });
+
+  server.listen(0, async () => {
+    const addr = server.address() as { port: number };
+
+    // Upgrade request should get 401
+    const res1 = await fetch(`http://127.0.0.1:${addr.port}/`, {
+      headers: {
+        Connection: "Upgrade",
+        Upgrade: "websocket",
+        "Sec-WebSocket-Version": "13",
+        "Sec-WebSocket-Key": "QUFBQUFBQUFBQUFBQUFBQQ==",
+      },
+    });
+    assertEquals(res1.status, 401);
+    await res1.body?.cancel();
+
+    // Normal request should still work
+    const res2 = await fetch(`http://127.0.0.1:${addr.port}/`);
+    assertEquals(await res2.text(), "ok");
+
+    server.close(() => resolve());
+  });
+
+  await promise;
+});