transport: validate http 200 status for responses

internal/transport/http2_client.go

@@ -1286,7 +1286,7 @@ func (t *http2Client) operateHeaders(frame *http2.MetaHeadersFrame) {
 		statusGen      *status.Status
 		httpStatusCode *int
 		httpStatusErr  string
-		rawStatus      string
+		rawStatusCode  *codes.Code
 		// headerError is set if an error is encountered while parsing the headers
 		headerError string
 	)
@@ -1308,7 +1308,14 @@ func (t *http2Client) operateHeaders(frame *http2.MetaHeadersFrame) {
 		case "grpc-encoding":
 			s.recvCompress = hf.Value
 		case "grpc-status":
-			rawStatus = hf.Value
+			code, err := strconv.ParseInt(hf.Value, 10, 32)
+			if err != nil {
+				se := status.New(codes.Internal, fmt.Sprintf("transport: malformed grpc-status: %v", err))
+				t.closeStream(s, se.Err(), true, http2.ErrCodeProtocol, se, nil, endStream)
+				return
+			}
+			c := codes.Code(uint32(code))
+			rawStatusCode = &c
 		case "grpc-message":
 			grpcMessage = decodeGrpcMessage(hf.Value)
 		case "grpc-status-details-bin":
@@ -1428,17 +1435,11 @@ func (t *http2Client) operateHeaders(frame *http2.MetaHeadersFrame) {
 	}
 
 	if statusGen == nil {
-		rawStatusCode := codes.Unknown
-		if rawStatus != "" {
-			code, err := strconv.ParseInt(rawStatus, 10, 32)
-			if err != nil {
-				se := status.New(codes.Internal, fmt.Sprintf("transport: malformed grpc-status: %v", err))
-				t.closeStream(s, se.Err(), true, http2.ErrCodeProtocol, se, nil, endStream)
-				return
-			}
-			rawStatusCode = codes.Code(uint32(code))
+		rsc := codes.Unknown
+		if rawStatusCode != nil {
+			rsc = *rawStatusCode
 		}
-		statusGen = status.New(rawStatusCode, grpcMessage)
+		statusGen = status.New(rsc, grpcMessage)
 	}
 
 	// if client received END_STREAM from server while stream was still active, send RST_STREAM

internal/transport/transport_test.go

@@ -1978,6 +1978,31 @@ func (s) TestClientHandshakeInfo(t *testing.T) {
 }
 
 func (s) TestClientDecodeHeaderStatusErr(t *testing.T) {
+	testStream := func() *Stream {
+		return &Stream{
+			done:       make(chan struct{}),
+			headerChan: make(chan struct{}),
+			buf: &recvBuffer{
+				c:  make(chan recvMsg),
+				mu: sync.Mutex{},
+			},
+		}
+	}
+
+	testClient := func(ts *Stream) *http2Client {
+		return &http2Client{
+			mu: sync.Mutex{},
+			activeStreams: map[uint32]*Stream{
+				0: ts,
+			},
+			controlBuf: &controlBuffer{
+				ch:   make(chan struct{}),
+				done: make(chan struct{}),
+				list: &itemList{},
+			},
+		}
+	}
+
 	for _, test := range []struct {
 		name string
 		// input
@@ -1993,12 +2018,6 @@ func (s) TestClientDecodeHeaderStatusErr(t *testing.T) {
 					{Name: "grpc-status", Value: "0"},
 					{Name: ":status", Value: "200"},
 				},
-				HeadersFrame: &http2.HeadersFrame{
-					FrameHeader: http2.FrameHeader{
-						StreamID: 0,
-						Flags:    http2.FlagHeadersEndStream,
-					},
-				},
 			},
 			// no error
 			wantStatus: status.New(codes.OK, ""),
@@ -2010,12 +2029,6 @@ func (s) TestClientDecodeHeaderStatusErr(t *testing.T) {
 					{Name: "grpc-status", Value: "0"},
 					{Name: ":status", Value: "200"},
 				},
-				HeadersFrame: &http2.HeadersFrame{
-					FrameHeader: http2.FrameHeader{
-						StreamID: 0,
-						Flags:    http2.FlagHeadersEndStream,
-					},
-				},
 			},
 			wantStatus: status.New(
 				codes.Unknown,
@@ -2030,12 +2043,6 @@ func (s) TestClientDecodeHeaderStatusErr(t *testing.T) {
 					{Name: "grpc-status", Value: "xxxx"},
 					{Name: ":status", Value: "200"},
 				},
-				HeadersFrame: &http2.HeadersFrame{
-					FrameHeader: http2.FrameHeader{
-						StreamID: 0,
-						Flags:    http2.FlagHeadersEndStream,
-					},
-				},
 			},
 			wantStatus: status.New(
 				codes.Internal,
@@ -2048,12 +2055,6 @@ func (s) TestClientDecodeHeaderStatusErr(t *testing.T) {
 				Fields: []hpack.HeaderField{
 					{Name: "content-type", Value: "application/json"},
 				},
-				HeadersFrame: &http2.HeadersFrame{
-					FrameHeader: http2.FrameHeader{
-						StreamID: 0,
-						Flags:    http2.FlagHeadersEndStream,
-					},
-				},
 			},
 			wantStatus: status.New(
 				codes.Internal,
@@ -2067,12 +2068,6 @@ func (s) TestClientDecodeHeaderStatusErr(t *testing.T) {
 					// No content type provided then fallback into handling http error.
 					{Name: ":status", Value: "xxxx"},
 				},
-				HeadersFrame: &http2.HeadersFrame{
-					FrameHeader: http2.FrameHeader{
-						StreamID: 0,
-						Flags:    http2.FlagHeadersEndStream,
-					},
-				},
 			},
 			wantStatus: status.New(
 				codes.Internal,
@@ -2084,12 +2079,6 @@ func (s) TestClientDecodeHeaderStatusErr(t *testing.T) {
 			metaHeaderFrame: &http2.MetaHeadersFrame{
 				Fields:    nil,
 				Truncated: true,
-				HeadersFrame: &http2.HeadersFrame{
-					FrameHeader: http2.FrameHeader{
-						StreamID: 0,
-						Flags:    http2.FlagHeadersEndStream,
-					},
-				},
 			},
 			wantStatus: status.New(
 				codes.Internal,
@@ -2104,11 +2093,6 @@ func (s) TestClientDecodeHeaderStatusErr(t *testing.T) {
 					{Name: "grpc-status", Value: "0"},
 					{Name: ":status", Value: "504"},
 				},
-				HeadersFrame: &http2.HeadersFrame{
-					FrameHeader: http2.FrameHeader{
-						StreamID: 0,
-					},
-				},
 			},
 			wantStatus: status.New(
 				codes.Unavailable,
@@ -2121,37 +2105,40 @@ func (s) TestClientDecodeHeaderStatusErr(t *testing.T) {
 				Fields: []hpack.HeaderField{
 					{Name: "content-type", Value: "application/grpc"},
 				},
-				HeadersFrame: &http2.HeadersFrame{
-					FrameHeader: http2.FrameHeader{
-						StreamID: 0,
-						Flags:    http2.FlagHeadersEndStream,
-					},
-				},
 			},
 			wantStatus: status.New(
 				codes.Internal,
 				"malformed header: missing HTTP status",
 			),
 		},
 	} {
+
 		t.Run(test.name, func(t *testing.T) {
-			ts := &Stream{
-				done:       make(chan struct{}),
-				headerChan: make(chan struct{}),
-				buf: &recvBuffer{
-					c:  make(chan recvMsg),
-					mu: sync.Mutex{},
+			ts := testStream()
+			s := testClient(ts)
+
+			test.metaHeaderFrame.HeadersFrame = &http2.HeadersFrame{
+				FrameHeader: http2.FrameHeader{
+					StreamID: 0,
 				},
 			}
-			s := &http2Client{
-				mu: sync.Mutex{},
-				activeStreams: map[uint32]*Stream{
-					0: ts,
-				},
-				controlBuf: &controlBuffer{
-					ch:   make(chan struct{}),
-					done: make(chan struct{}),
-					list: &itemList{},
+
+			s.operateHeaders(test.metaHeaderFrame)
+
+			got := ts.status
+			want := test.wantStatus
+			if got.Code() != want.Code() || got.Message() != want.Message() {
+				t.Fatalf("operateHeaders(%v); status = \ngot: %s\nwant: %s", test.metaHeaderFrame, got, want)
+			}
+		})
+		t.Run(fmt.Sprintf("%s-end_stream", test.name), func(t *testing.T) {
+			ts := testStream()
+			s := testClient(ts)
+
+			test.metaHeaderFrame.HeadersFrame = &http2.HeadersFrame{
+				FrameHeader: http2.FrameHeader{
+					StreamID: 0,
+					Flags:    http2.FlagHeadersEndStream,
 				},
 			}
 

test/end2end_test.go

@@ -7263,7 +7263,7 @@ func (s) TestHTTPHeaderFrameErrorHandlingInitialHeader(t *testing.T) {
 				"content-type", "application/grpc",
 				"grpc-status", "abc",
 			},
-			errCode: codes.Unavailable,
+			errCode: codes.Internal,
 		},
 		{
 			// Malformed grpc-tags-bin field.