transport: http2 server must validates header list size when early aborting stream

internal/transport/controlbuf.go

@@ -150,11 +150,12 @@ type cleanupStream struct {
 func (c *cleanupStream) isTransportResponseFrame() bool { return c.rst } // Results in a RST_STREAM
 
 type earlyAbortStream struct {
-	httpStatus     uint32
-	streamID       uint32
-	contentSubtype string
-	status         *status.Status
-	rst            bool
+	httpStatus            uint32
+	streamID              uint32
+	contentSubtype        string
+	status                *status.Status
+	rst                   bool
+	maxSendHeaderListSize *uint32
 }
 
 func (*earlyAbortStream) isTransportResponseFrame() bool { return false }
@@ -866,6 +867,13 @@ func (l *loopyWriter) earlyAbortStreamHandler(eas *earlyAbortStream) error {
 		}
 	}
 
+	if !checkForHeaderListSize(headerFields, eas.maxSendHeaderListSize) {
+		if l.logger.V(logLevel) {
+			l.logger.Infof("Header list size to send violates the maximum size (%d bytes) set by client", *eas.maxSendHeaderListSize)
+		}
+		return l.framer.fr.WriteRSTStream(eas.streamID, http2.ErrCodeInternal)
+	}
+
 	if err := l.writeHeader(eas.streamID, true, headerFields, nil); err != nil {
 		return err
 	}

internal/transport/http2_server.go

@@ -480,11 +480,12 @@ func (t *http2Server) operateHeaders(ctx context.Context, frame *http2.MetaHeade
 			t.logger.Infof("Aborting the stream early: %v", errMsg)
 		}
 		t.controlBuf.put(&earlyAbortStream{
-			httpStatus:     http.StatusBadRequest,
-			streamID:       streamID,
-			contentSubtype: s.contentSubtype,
-			status:         status.New(codes.Internal, errMsg),
-			rst:            !frame.StreamEnded(),
+			httpStatus:            http.StatusBadRequest,
+			streamID:              streamID,
+			contentSubtype:        s.contentSubtype,
+			status:                status.New(codes.Internal, errMsg),
+			rst:                   !frame.StreamEnded(),
+			maxSendHeaderListSize: t.maxSendHeaderListSize,
 		})
 		return nil
 	}
@@ -500,21 +501,23 @@ func (t *http2Server) operateHeaders(ctx context.Context, frame *http2.MetaHeade
 	}
 	if !isGRPC {
 		t.controlBuf.put(&earlyAbortStream{
-			httpStatus:     http.StatusUnsupportedMediaType,
-			streamID:       streamID,
-			contentSubtype: s.contentSubtype,
-			status:         status.Newf(codes.InvalidArgument, "invalid gRPC request content-type %q", contentType),
-			rst:            !frame.StreamEnded(),
+			httpStatus:            http.StatusUnsupportedMediaType,
+			streamID:              streamID,
+			contentSubtype:        s.contentSubtype,
+			status:                status.Newf(codes.InvalidArgument, "invalid gRPC request content-type %q", contentType),
+			rst:                   !frame.StreamEnded(),
+			maxSendHeaderListSize: t.maxSendHeaderListSize,
 		})
 		return nil
 	}
 	if headerError != nil {
 		t.controlBuf.put(&earlyAbortStream{
-			httpStatus:     http.StatusBadRequest,
-			streamID:       streamID,
-			contentSubtype: s.contentSubtype,
-			status:         headerError,
-			rst:            !frame.StreamEnded(),
+			httpStatus:            http.StatusBadRequest,
+			streamID:              streamID,
+			contentSubtype:        s.contentSubtype,
+			status:                headerError,
+			rst:                   !frame.StreamEnded(),
+			maxSendHeaderListSize: t.maxSendHeaderListSize,
 		})
 		return nil
 	}
@@ -570,11 +573,12 @@ func (t *http2Server) operateHeaders(ctx context.Context, frame *http2.MetaHeade
 			t.logger.Infof("Aborting the stream early: %v", errMsg)
 		}
 		t.controlBuf.put(&earlyAbortStream{
-			httpStatus:     http.StatusMethodNotAllowed,
-			streamID:       streamID,
-			contentSubtype: s.contentSubtype,
-			status:         status.New(codes.Internal, errMsg),
-			rst:            !frame.StreamEnded(),
+			httpStatus:            http.StatusMethodNotAllowed,
+			streamID:              streamID,
+			contentSubtype:        s.contentSubtype,
+			status:                status.New(codes.Internal, errMsg),
+			rst:                   !frame.StreamEnded(),
+			maxSendHeaderListSize: t.maxSendHeaderListSize,
 		})
 		s.cancel()
 		return nil
@@ -591,11 +595,12 @@ func (t *http2Server) operateHeaders(ctx context.Context, frame *http2.MetaHeade
 				stat = status.New(codes.PermissionDenied, err.Error())
 			}
 			t.controlBuf.put(&earlyAbortStream{
-				httpStatus:     http.StatusOK,
-				streamID:       s.id,
-				contentSubtype: s.contentSubtype,
-				status:         stat,
-				rst:            !frame.StreamEnded(),
+				httpStatus:            http.StatusOK,
+				streamID:              s.id,
+				contentSubtype:        s.contentSubtype,
+				status:                stat,
+				rst:                   !frame.StreamEnded(),
+				maxSendHeaderListSize: t.maxSendHeaderListSize,
 			})
 			return nil
 		}
@@ -605,11 +610,12 @@ func (t *http2Server) operateHeaders(ctx context.Context, frame *http2.MetaHeade
 		t.mu.Unlock()
 		// Early abort in case the timeout was zero or so low it already fired.
 		t.controlBuf.put(&earlyAbortStream{
-			httpStatus:     http.StatusOK,
-			streamID:       s.id,
-			contentSubtype: s.contentSubtype,
-			status:         status.New(codes.DeadlineExceeded, context.DeadlineExceeded.Error()),
-			rst:            !frame.StreamEnded(),
+			httpStatus:            http.StatusOK,
+			streamID:              s.id,
+			contentSubtype:        s.contentSubtype,
+			status:                status.New(codes.DeadlineExceeded, context.DeadlineExceeded.Error()),
+			rst:                   !frame.StreamEnded(),
+			maxSendHeaderListSize: t.maxSendHeaderListSize,
 		})
 		return nil
 	}
@@ -969,23 +975,32 @@ func appendHeaderFieldsFromMD(headerFields []hpack.HeaderField, md metadata.MD)
 	return headerFields
 }
 
-func (t *http2Server) checkForHeaderListSize(it any) bool {
-	if t.maxSendHeaderListSize == nil {
+// checkForHeaderListSize checks if the header list size exceeds the limit set
+// by the peer. It returns false if the limit is exceeded.
+func checkForHeaderListSize(hf []hpack.HeaderField, maxSendHeaderListSize *uint32) bool {
+	if maxSendHeaderListSize == nil {
 		return true
 	}
-	hdrFrame := it.(*headerFrame)
 	var sz int64
-	for _, f := range hdrFrame.hf {
-		if sz += int64(f.Size()); sz > int64(*t.maxSendHeaderListSize) {
-			if t.logger.V(logLevel) {
-				t.logger.Infof("Header list size to send violates the maximum size (%d bytes) set by client", *t.maxSendHeaderListSize)
-			}
+	for _, f := range hf {
+		if sz += int64(f.Size()); sz > int64(*maxSendHeaderListSize) {
 			return false
 		}
 	}
 	return true
 }
 
+func (t *http2Server) checkForHeaderListSize(it any) bool {
+	hdrFrame := it.(*headerFrame)
+	if !checkForHeaderListSize(hdrFrame.hf, t.maxSendHeaderListSize) {
+		if t.logger.V(logLevel) {
+			t.logger.Infof("Header list size to send violates the maximum size (%d bytes) set by client", *t.maxSendHeaderListSize)
+		}
+		return false
+	}
+	return true
+}
+
 func (t *http2Server) streamContextErr(s *ServerStream) error {
 	select {
 	case <-t.done:

test/end2end_test.go

@@ -6134,6 +6134,43 @@ func testClientMaxHeaderListSizeServerIntentionalViolation(t *testing.T, e env)
 	}
 }
 
+func (s) TestEarlyAbortStreamHeaderListSizeCheck(t *testing.T) {
+	lis, err := net.Listen("tcp", "localhost:0")
+	if err != nil {
+		t.Fatalf("Failed to listen: %v", err)
+	}
+	s := grpc.NewServer()
+	defer s.Stop()
+	go s.Serve(lis)
+
+	conn, err := net.DialTimeout("tcp", lis.Addr().String(), defaultTestTimeout)
+	if err != nil {
+		t.Fatalf("Failed to dial: %v", err)
+	}
+	defer conn.Close()
+	st := newServerTesterFromConn(t, conn)
+
+	// Set a very small MaxHeaderListSize that any response headers would violate.
+	st.greetWithSettings(http2.Setting{ID: http2.SettingMaxHeaderListSize, Val: 1})
+
+	// Send a request with an invalid content-type to trigger early abort.
+	st.writeHeaders(http2.HeadersFrameParam{
+		StreamID: 1,
+		BlockFragment: st.encodeHeader(
+			":method", "POST",
+			":path", "/grpc.testing.TestService/UnaryCall",
+			"content-type", "text/plain", // Invalid content-type to trigger early abort
+			"te", "trailers",
+		),
+		EndStream:  true,
+		EndHeaders: true,
+	})
+
+	// We should receive a RST_STREAM with ErrCodeInternal because the response
+	// headers exceed the MaxHeaderListSize limit.
+	st.wantRSTStream(http2.ErrCodeInternal)
+}
+
 func (s) TestNetPipeConn(t *testing.T) {
 	// This test will block indefinitely if grpc writes both client and server
 	// prefaces without either reading from the Conn.

test/servertester.go

@@ -91,8 +91,19 @@ func (st *serverTester) readFrame() (http2.Frame, error) {
 // greet initiates the client's HTTP/2 connection into a state where
 // frames may be sent.
 func (st *serverTester) greet() {
+	st.greetWithSettings()
+}
+
+// greetWithSettings initiates the client's HTTP/2 connection with custom settings.
+func (st *serverTester) greetWithSettings(settings ...http2.Setting) {
 	st.writePreface()
-	st.writeInitialSettings()
+	if len(settings) > 0 {
+		if err := st.fr.WriteSettings(settings...); err != nil {
+			st.t.Fatalf("Error writing initial SETTINGS frame from client to server: %v", err)
+		}
+	} else {
+		st.writeInitialSettings()
+	}
 	st.wantSettings()
 	st.writeSettingsAck()
 	for {