node:http2: enforce RFC 9113 message framing on received responses

src/js/node/http2.ts

@@ -376,7 +376,6 @@ const kReceivedGoaway = Symbol("receivedGoaway");
 const kGoawayCode = Symbol("goawayCode");
 const kReleaseUnannouncedStream = Symbol("releaseUnannouncedStream");
 const kGoawaySent = Symbol("goawaySent");
-const kSocketTeardown = Symbol("socketTeardown");
 const kMaxStreams = 2 ** 32 - 1;
 const kMaxUint32 = 4294967295;
 const kMaxInt = 4294967295;
@@ -3159,6 +3158,9 @@ function emitConnectNT(self, socket) {
 function destroyWithInvalidSessionNT(stream) {
   if (!stream.destroyed) stream.destroy($ERR_HTTP2_INVALID_SESSION());
 }
+function destroyWithGoawaySessionNT(stream) {
+  if (!stream.destroyed) stream.destroy($ERR_HTTP2_GOAWAY_SESSION());
+}
 function destroyIfNotDestroyedNT(target) {
   if (!target.destroyed) target.destroy();
 }
@@ -4101,20 +4103,18 @@ function streamSocketClosed(stream: Http2Stream) {
   }
 }
 // A stream whose session was close()d before the socket finished connecting never reached the
-// peer; node destroys it with ERR_HTTP2_GOAWAY_SESSION (no $ERR intrinsic exists for this code).
+// peer; node destroys it with ERR_HTTP2_GOAWAY_SESSION.
 function rejectStreamAboveGoawayLastId(lastStreamId: number, stream: Http2Stream) {
   if (typeof stream?.id === "number" && stream.id > lastStreamId) {
     streamRejectedByGoawaySession(stream);
   }
 }
 function streamRejectedByGoawaySession(stream: Http2Stream) {
   if (!stream.destroyed) {
-    const err = new Error("New streams cannot be created after receiving a GOAWAY");
-    err.code = "ERR_HTTP2_GOAWAY_SESSION";
     // nghttp2 closes unprocessed streams with REFUSED_STREAM, the signal clients (grpc) treat
     // as safely retryable on a fresh connection.
     stream.rstCode = constants.NGHTTP2_REFUSED_STREAM;
-    stream.destroy(err);
+    stream.destroy($ERR_HTTP2_GOAWAY_SESSION());
   }
 }
 class ClientHttp2Session extends Http2Session {
@@ -4388,8 +4388,8 @@ class ClientHttp2Session extends Http2Session {
       if (self.destroyed) return;
       self[kGoawayCode] = errorCode;
       // node: once a GOAWAY is received, new streams cannot be created on this session -
-      // request() throws ERR_HTTP2_GOAWAY_SESSION (clients like grpc rely on the throw to
-      // fail over to a fresh connection).
+      // request() returns a stream that errors with ERR_HTTP2_GOAWAY_SESSION (clients like
+      // grpc rely on that error class to fail over to a fresh connection).
       self[kReceivedGoaway] = true;
       self.emit("goaway", errorCode, lastStreamId, opaqueData || Buffer.allocUnsafe(0));
       // node: streams the peer did not process (id above the GOAWAY's lastStreamId) are
@@ -4417,7 +4417,6 @@ class ClientHttp2Session extends Http2Session {
     },
     end(self: ClientHttp2Session, errorCode: number, lastStreamId: number, opaqueData: Buffer) {
       if (!self) return;
-      self[kSocketTeardown] = true;
       self.destroy();
     },
     altsvc(self: ClientHttp2Session, origin: string, value: string, streamId: number) {
@@ -4503,14 +4502,10 @@ class ClientHttp2Session extends Http2Session {
       parser.detach();
       this.#parser = null;
     }
-    // Socket-driven teardown: node's destroyed flag flips asynchronously in this path, so a
-    // request() racing the teardown returns a stream that errors instead of throwing.
-    this[kSocketTeardown] = true;
     this.destroy(err, NGHTTP2_NO_ERROR);
     this[bunHTTP2Socket] = null;
   }
   #onError(error: Error) {
-    this[kSocketTeardown] = true;
     this[bunHTTP2Socket] = null;
     if (this.#closed) {
       this.destroy();
@@ -4900,30 +4895,19 @@ class ClientHttp2Session extends Http2Session {
     // throws before that point must not decrement.
     let connectionsCounted = false;
     try {
-      // node: a destroyed session reports INVALID_SESSION; a closed (GOAWAY) one reports
-      // GOAWAY_SESSION. When the destruction came from the socket
-      // tearing down (not an explicit destroy()), node's destroyed flag flips asynchronously -
-      // a racing request() must not throw, it returns a stream that errors.
+      // node: request() on a destroyed or closed session never throws. It returns a stream that
+      // is destroyed on the next tick with the matching error: ERR_HTTP2_INVALID_SESSION for a
+      // destroyed session, ERR_HTTP2_GOAWAY_SESSION for a closed one or after a received GOAWAY
+      // (verified node v26.3.0).
       if (this.destroyed) {
-        if (this[kSocketTeardown]) {
-          const req = new ClientHttp2Stream(undefined, this, headers);
-          process.nextTick(destroyWithInvalidSessionNT, req);
-          return req;
-        }
-        throw $ERR_HTTP2_INVALID_SESSION();
-      }
-      if (this[kReceivedGoaway]) {
-        const err = new Error("New streams cannot be created after receiving a GOAWAY");
-        err.code = "ERR_HTTP2_GOAWAY_SESSION";
-        throw err;
+        const req = new ClientHttp2Stream(undefined, this, headers);
+        process.nextTick(destroyWithInvalidSessionNT, req);
+        return req;
       }
-      if (this.closed) {
-        // node: a closed (close() called / GOAWAY pending) session reports
-        // ERR_HTTP2_GOAWAY_SESSION on the stream (verified node v26.3.0); the test
-        // contract accepts a synchronous throw of the same error.
-        const err = new Error("New streams cannot be created after receiving a GOAWAY");
-        err.code = "ERR_HTTP2_GOAWAY_SESSION";
-        throw err;
+      if (this.closed || this[kReceivedGoaway]) {
+        const req = new ClientHttp2Stream(undefined, this, headers);
+        process.nextTick(destroyWithGoawaySessionNT, req);
+        return req;
       }
 
       if (this.sentTrailers) {

src/runtime/api/bun/h2/connection.rs

@@ -25,6 +25,10 @@ pub struct Stream {
     pub content_length: Option<u64>,
     /// DATA payload bytes (padding excluded) received so far.
     pub recv_body_bytes: u64,
+    /// `:method` of the request this stream's response answers, recorded from a PUSH_PROMISE
+    /// header block the engine decoded itself (nghttp2's `nghttp2_http_record_request_method`).
+    /// `Other` for a request the embedder encoded locally; the `Sink` shim reports those.
+    pub local_method: LocalRequestMethod,
 }
 
 impl Stream {
@@ -36,10 +40,51 @@ impl Stream {
             recv_final_headers: false,
             content_length: None,
             recv_body_bytes: 0,
+            local_method: LocalRequestMethod::Other,
         }
     }
 }
 
+/// `:method` of the request a response answers. Only HEAD and CONNECT change whether that
+/// response may carry a `content-length`-governed body (RFC 9113 §8.1.1).
+#[derive(Clone, Copy, PartialEq, Eq, Debug, Default)]
+pub enum LocalRequestMethod {
+    #[default]
+    Other,
+    /// HEAD: the response never carries a body, whatever `content-length` it declares.
+    Head,
+    /// CONNECT: DATA on the response is a tunnel, not a `content-length`-governed body. nghttp2
+    /// exempts every CONNECT response, not only the 2xx that opens the tunnel; node matches.
+    Connect,
+}
+
+impl LocalRequestMethod {
+    /// Classify an outbound `:method` the way nghttp2_http_record_request_method does.
+    /// Methods are case-sensitive.
+    pub fn from_method(value: &[u8]) -> Self {
+        match value {
+            b"HEAD" => LocalRequestMethod::Head,
+            b"CONNECT" => LocalRequestMethod::Connect,
+            _ => LocalRequestMethod::Other,
+        }
+    }
+}
+
+/// RFC 9113 §8.3.2: `:status` is exactly three ASCII digits, >= 100, and never 101 (Switching
+/// Protocols has no HTTP/2 meaning). Returns `None` for an invalid value.
+fn parse_status(value: &[u8]) -> Option<u16> {
+    if value.len() != 3 || !value.iter().all(u8::is_ascii_digit) {
+        return None;
+    }
+    let code = (u16::from(value[0] - b'0')) * 100
+        + (u16::from(value[1] - b'0')) * 10
+        + u16::from(value[2] - b'0');
+    if code < 100 || code == 101 {
+        return None;
+    }
+    Some(code)
+}
+
 /// RFC 9110 §8.6: `content-length` is 1*DIGIT. Anything else, or a value that does not fit
 /// in a u64, is rejected.
 fn parse_content_length(value: &[u8]) -> Option<u64> {
@@ -143,6 +188,13 @@ pub trait Sink {
     fn is_local_stream(&self, _stream_id: u32) -> bool {
         false
     }
+    /// Transition shim (same reason as `is_local_stream`): the `:method` the embedder sent on
+    /// `stream_id`. Consulted only for a client's final response header block, because HEAD and
+    /// CONNECT change the content-length semantics of the response (RFC 9113 §8.1.1) and the
+    /// engine never saw the outbound request HEADERS.
+    fn local_request_method(&self, _stream_id: u32) -> LocalRequestMethod {
+        LocalRequestMethod::Other
+    }
 }
 
 pub struct Connection {
@@ -482,7 +534,12 @@ impl Connection {
         let mut evict = std::mem::take(&mut self.evict_buf);
         evict.clear();
         for (id, s) in self.streams.iter() {
-            if s.state == State::Closed {
+            // A Closed stream whose header block is still being reassembled across CONTINUATION
+            // frames (§4.3) must survive to finish_header_block: the entry carries the message
+            // state that categorizes the block and validates it (a pushed request's `:method`).
+            if s.state == State::Closed
+                && (self.continuation_stream == 0 || *id != self.header_target)
+            {
                 evict.push(*id);
             }
         }
@@ -875,10 +932,23 @@ impl Connection {
                 .streams
                 .get(&target)
                 .is_some_and(|s| s.recv_final_headers);
+        // §8.3: the pseudo-header fields a block of this category may carry, as a bitmask over the
+        // `bit` values assigned below. A request block is a server's inbound block or a client's
+        // PUSH_PROMISE block (the promised request); everything else inbound is a response. §8.1:
+        // a trailer section carries none.
+        let allowed_pseudo: u8 = if is_trailer {
+            0
+        } else if self.is_server || push_parent != 0 {
+            1 | 2 | 4 | 8 | 32 // :method :scheme :authority :path :protocol
+        } else {
+            16 // :status
+        };
         let mut rejected = false;
         let mut malformed = is_trailer && !self.header_end_stream;
         let mut seen_regular = false;
         let mut seen_pseudo: u8 = 0;
+        let mut status: Option<u16> = None;
+        let mut block_method = LocalRequestMethod::Other;
         let mut informational = false;
         let mut content_length: Option<u64> = None;
         let mut connect = false;
@@ -921,26 +991,28 @@ impl Connection {
                                 b"protocol" => 32,
                                 _ => 64,
                             };
-                            // 8.3.1: requests never carry :status - a server seeing it inbound is
-                            // a malformed block. (The client direction also constrains pseudo
-                            // headers, but inbound PUSH_PROMISE blocks legitimately carry request
-                            // pseudo-headers, so that check needs the push context first.)
-                            let wrong_direction = self.is_server && rest == b"status";
+                            // §8.3: a pseudo-header following a regular field, an unknown or
+                            // repeated pseudo-header, or one outside the set its block's category
+                            // defines (`:status` in a request, a request pseudo-header in a
+                            // response, any pseudo-header in a trailer section) is malformed.
                             if seen_regular
                                 || bit == 64
                                 || (seen_pseudo & bit) != 0
-                                || wrong_direction
-                                || is_trailer
+                                || (bit & allowed_pseudo) == 0
                             {
                                 malformed = true;
-                            }
-                            if rest == b"status" && value_b.len() == 3 && value_b[0] == b'1' {
-                                informational = true;
+                            } else if bit == 16 {
+                                // §8.3.2: three digits, >= 100, never 101.
+                                status = parse_status(value_b);
+                                match status {
+                                    None => malformed = true,
+                                    Some(code) => informational = code / 100 == 1,
+                                }
+                            } else if bit == 1 {
+                                block_method = LocalRequestMethod::from_method(value_b);
+                                connect = value_b == b"CONNECT";
                             }
                             seen_pseudo |= bit;
-                            if rest == b"method" && value_b == b"CONNECT" {
-                                connect = true;
-                            }
                         } else {
                             seen_regular = true;
                             match name_b {
@@ -996,11 +1068,64 @@ impl Connection {
             sink.on_stream_reset(target, ErrorCode::StreamClosed.as_u32());
             return false;
         }
-        if push_parent == 0 && self.is_server && !malformed && !rejected {
-            if let Some(s) = self.streams.get_mut(&target) {
-                if !connect && s.content_length.is_none() {
-                    s.content_length = content_length;
+        // RFC 9113 §8.1.1/§8.3 message-body rules, the equivalent of nghttp2's
+        // nghttp2_http_on_{request,response,trailer}_headers. `declared` is what this block says
+        // the message body's length is; a block that does not bind it (an interim response, a
+        // tunnel, a trailer section) only has to meet what the header section already declared.
+        if push_parent != 0 {
+            // A PUSH_PROMISE block is the promised stream's request: only its `:method` binds,
+            // deciding whether the pushed response may carry a body (the stream's own first
+            // inbound HEADERS is that response). nghttp2_http_record_request_method.
+            if !malformed
+                && !rejected
+                && let Some(s) = self.streams.get_mut(&target)
+            {
+                s.local_method = block_method;
+            }
+        } else if !malformed && !rejected {
+            let mut binds = true;
+            let declared: Option<u64> = if is_trailer {
+                binds = false;
+                None
+            } else if self.is_server {
+                // §8.5: a (non-extended) CONNECT request's DATA is a tunnel, not a body.
+                if connect { None } else { content_length }
+            } else if (seen_pseudo & 16) == 0 {
+                // §8.3.2: a response header section carries exactly one `:status` (its validity
+                // was checked as it was decoded).
+                malformed = true;
+                binds = false;
+                None
+            } else if informational {
+                // An interim response: the final header section follows on this stream, so this
+                // block cannot also end it, and its fields don't bind the message.
+                if self.header_end_stream {
+                    malformed = true;
+                }
+                binds = false;
+                None
+            } else {
+                // §8.1.1: a HEAD response and a 204/304 carry no body; a CONNECT response is a
+                // tunnel (nghttp2's expect_response_body()). The request's `:method` comes from
+                // the promised request's block for a pushed stream, and from the Sink shim (the
+                // embedder's legacy encoder sent it) for a locally-initiated one.
+                let method = match self.streams.get(&target).map(|s| s.local_method) {
+                    Some(m) if m != LocalRequestMethod::Other => m,
+                    _ => sink.local_request_method(target),
+                };
+                match status {
+                    Some(204) | Some(304) => Some(0),
+                    _ if method == LocalRequestMethod::Head => Some(0),
+                    _ if method == LocalRequestMethod::Connect => None,
+                    _ => content_length,
+                }
+            };
+            if !malformed && let Some(s) = self.streams.get_mut(&target) {
+                if binds && s.content_length.is_none() {
+                    s.content_length = declared;
                 }
+                // The declared length must be met by the DATA actually received once this block
+                // ends the stream; `enforce_content_length` covers an END_STREAM on DATA.
                 if self.header_end_stream
                     && s.content_length
                         .is_some_and(|declared| declared != s.recv_body_bytes)
@@ -1120,6 +1245,19 @@ impl Connection {
                         discard = true;
                     } else {
                         st.recv_body_bytes = st.recv_body_bytes.saturating_add(data_total as u64);
+                        // §8.1.1: DATA beyond the declared content-length is a stream
+                        // PROTOCOL_ERROR, and the excess never reaches the application.
+                        if st
+                            .content_length
+                            .is_some_and(|declared| st.recv_body_bytes > declared)
+                        {
+                            self.send_rst_stream(sink, hdr.stream_id, ErrorCode::ProtocolError);
+                            if let Some(st2) = self.streams.get_mut(&hdr.stream_id) {
+                                st2.state = State::Closed;
+                            }
+                            sink.on_stream_reset(hdr.stream_id, ErrorCode::ProtocolError.as_u32());
+                            discard = true;
+                        }
                     }
                 }
             }
@@ -1236,7 +1374,15 @@ impl Connection {
                         DataDecision::Rst(ErrorCode::FlowControlError)
                     } else {
                         s.recv_body_bytes = s.recv_body_bytes.saturating_add((end - off) as u64);
-                        DataDecision::Deliver(0)
+                        // §8.1.1: DATA beyond the declared content-length is a stream
+                        // PROTOCOL_ERROR; the excess never reaches the application.
+                        if s.content_length
+                            .is_some_and(|declared| s.recv_body_bytes > declared)
+                        {
+                            DataDecision::Rst(ErrorCode::ProtocolError)
+                        } else {
+                            DataDecision::Deliver(0)
+                        }
                     }
                 }
             }
@@ -1284,13 +1430,11 @@ impl Connection {
         false
     }
 
-    /// RFC 9113 §8.1.1: once END_STREAM arrives, a request whose received DATA total contradicts
+    /// RFC 9113 §8.1.1: once END_STREAM arrives, a message whose received DATA total contradicts
     /// its declared `content-length` is malformed. Resets the stream with PROTOCOL_ERROR instead
-    /// of signalling end-of-stream and returns true if it did so.
+    /// of signalling end-of-stream and returns true if it did so. Applies in both directions: a
+    /// truncated response is the only signal a client gets that it did not receive the whole body.
     fn enforce_content_length(&mut self, sink: &impl Sink, stream_id: u32) -> bool {
-        if !self.is_server {
-            return false;
-        }
         let mismatch = self.streams.get(&stream_id).is_some_and(|s| {
             s.content_length
                 .is_some_and(|declared| declared != s.recv_body_bytes)