secfix

Author: eshitachandwani

internal/transport/http2_server.go

@@ -531,6 +531,7 @@ func (t *http2Server) operateHeaders(ctx context.Context, frame *http2.MetaHeade
 	if frame.StreamEnded() {
 		// s is just created by the caller. No lock needed.
 		s.state = streamReadDone
+		s.write(recvMsg{err: io.EOF})
 	}
 	if timeoutSet {
 		s.ctx, s.cancel = context.WithTimeout(ctx, timeout)

test/end2end_test.go

@@ -4765,14 +4765,20 @@ func testClientInitialHeaderEndStream(t *testing.T, e env) {
 	te := newTest(t, e)
 	ts := &funcServer{streamingInputCall: func(stream testgrpc.TestService_StreamingInputCallServer) error {
 		defer close(handlerDone)
-		// Block on serverTester receiving RST_STREAM. This ensures server has closed
-		// stream before stream.Recv().
+		// Block on serverTester receiving RST_STREAM. This ensures server has
+		// closed stream before stream.Recv().
 		<-frameCheckingDone
-		data, err := stream.Recv()
-		if err == nil {
-			t.Errorf("unexpected data received in func server method: '%v'", data)
+		// Depending on whether the context cancellation (due to the illegal data
+		// RST_STREAM) or the buffered EOF (from the initial HEADERS END_STREAM) is
+		// selected first in recvBufferReader, stream.Recv() can return either
+		// io.EOF or Canceled.
+		if _, err := stream.Recv(); err != io.EOF && status.Code(err) != codes.Canceled {
+			t.Errorf("expected EOF or canceled error, instead received '%v'", err)
+		}
+		if err := stream.SendMsg(nil); err == nil {
+			t.Error("expected error sending message on stream after stream closed due to illegal data")
 		} else if status.Code(err) != codes.Canceled {
-			t.Errorf("expected canceled error, instead received '%v'", err)
+			t.Errorf("expected cancel error, instead received '%v'", err)
 		}
 		return nil
 	}}

test/transport_test.go

@@ -302,3 +302,56 @@ func (s) TestCancelWhileServerWaitingForFlowControl(t *testing.T) {
 		t.Fatalf("Failed to read from the stream: %v", err)
 	}
 }
+
+// Tests that when a client sends a HEADERS frame with EndStream=true, the
+// server-side stream receives an io.EOF on Recv() and does not hang waiting
+// for data frames.
+func (s) TestHeadersEndStreamNoHang(t *testing.T) {
+	receivedErr := make(chan error, 1)
+	ss := &stubserver.StubServer{
+		FullDuplexCallF: func(stream testgrpc.TestService_FullDuplexCallServer) error {
+			_, err := stream.Recv()
+			receivedErr <- err
+			return nil
+		},
+	}
+	if err := ss.Start(nil); err != nil {
+		t.Fatalf("Error starting endpoint server: %v", err)
+	}
+	defer ss.Stop()
+
+	conn, err := net.DialTimeout("tcp", ss.Address, defaultTestTimeout)
+	if err != nil {
+		t.Fatalf("Failed to dial: %v", err)
+	}
+	defer conn.Close()
+
+	st := newServerTesterFromConn(t, conn)
+	st.greet()
+
+	// Send HEADERS with EndStream = true and no grpc-timeout header.
+	st.writeHeaders(http2.HeadersFrameParam{
+		StreamID: 1,
+		BlockFragment: st.encodeHeader(
+			":method", "POST",
+			":path", "/grpc.testing.TestService/FullDuplexCall",
+			":authority", "localhost",
+			"content-type", "application/grpc",
+			"te", "trailers",
+		),
+		EndStream:  true,
+		EndHeaders: true,
+	})
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	select {
+	case err := <-receivedErr:
+		if err != io.EOF {
+			t.Fatalf("Streaming handler expected io.EOF, got %v", err)
+		}
+	case <-ctx.Done():
+		t.Fatalf("Timed out waiting for Recv() on the server to complete: %v", ctx.Err())
+	}
+}