Drain stream and error on trailing data

connect_ext_test.go

@@ -2138,6 +2138,20 @@ func TestStreamUnexpectedEOF(t *testing.T) {
 		},
 		expectCode: connect.CodeInvalidArgument,
 		expectMsg:  "invalid_argument: protocol error: incomplete envelope: unexpected EOF",
+	}, {
+		name: "stream_excess_eof",
+		handler: func(responseWriter http.ResponseWriter, request *http.Request) {
+			_, _ = responseWriter.Write(head[:])
+			_, _ = responseWriter.Write(payload)
+			// Write EOF
+			_, _ = responseWriter.Write([]byte{2, 0, 0, 0, 2})
+			_, _ = responseWriter.Write([]byte("{}"))
+			// Excess payload
+			_, _ = responseWriter.Write(head[:])
+			_, _ = responseWriter.Write(payload)
+		},
+		expectCode: connect.CodeUnknown,
+		expectMsg:  fmt.Sprintf("unknown: corrupt response: %d extra bytes after end of stream", len(payload)+len(head)),
 	}}
 	for _, testcase := range testcases {
 		testcaseMux[t.Name()+"/"+testcase.name] = testcase.handler

duplex_http_call.go

@@ -179,7 +179,7 @@ func (d *duplexHTTPCall) CloseRead() error {
 	if d.response == nil {
 		return nil
 	}
-	if err := discard(d.response.Body); err != nil {
+	if _, err := discard(d.response.Body); err != nil {
 		_ = d.response.Body.Close()
 		return wrapIfRSTError(err)
 	}

envelope.go

@@ -209,6 +209,12 @@ func (r *envelopeReader) Unmarshal(message any) *Error {
 			Data:  bytes.NewBuffer(copiedData),
 			Flags: env.Flags,
 		}
+		// Drain the rest of the stream to ensure there is no extra data.
+		if n, err := discard(r.reader); err != nil {
+			return errorf(CodeUnknown, "corrupt response: I/O error after end-stream message: %w", err)
+		} else if n > 0 {
+			return errorf(CodeUnknown, "corrupt response: %d extra bytes after end of stream", n)
+		}
 		return errSpecialEnvelope
 	}
 

protocol.go

@@ -283,16 +283,14 @@ func isCommaOrSpace(c rune) bool {
 	return c == ',' || c == ' '
 }
 
-func discard(reader io.Reader) error {
+func discard(reader io.Reader) (int64, error) {
 	if lr, ok := reader.(*io.LimitedReader); ok {
-		_, err := io.Copy(io.Discard, lr)
-		return err
+		return io.Copy(io.Discard, lr)
 	}
 	// We don't want to get stuck throwing data away forever, so limit how much
 	// we're willing to do here.
 	lr := &io.LimitedReader{R: reader, N: discardLimit}
-	_, err := io.Copy(io.Discard, lr)
-	return err
+	return io.Copy(io.Discard, lr)
 }
 
 // negotiateCompression determines and validates the request compression and

protocol_grpc.go

@@ -325,8 +325,6 @@ func (g *grpcClient) NewConn(
 		}
 	} else {
 		conn.readTrailers = func(_ *grpcUnmarshaler, call *duplexHTTPCall) http.Header {
-			// To access HTTP trailers, we need to read the body to EOF.
-			_ = discard(call)
 			return call.ResponseTrailer()
 		}
 	}