proxy: reproduce and fix weird HTTP/2 error

proxy/proxy.go

@@ -24,6 +24,8 @@ const (
 	// "GET /availability/v1/btc.json HTTP/1.1" "" "Mozilla/5.0 ..."
 	formatPattern  = "- - \"%s %s %s\" \"%s\" \"%s\""
 	hdrContentType = "Content-Type"
+	hdrGrpcStatus  = "Grpc-Status"
+	hdrGrpcMessage = "Grpc-Message"
 	hdrTypeGrpc    = "application/grpc"
 )
 
@@ -170,7 +172,7 @@ func (p *Proxy) UpdateServices(services []*Service) error {
 
 	p.proxyBackend = &httputil.ReverseProxy{
 		Director:  p.director,
-		Transport: transport,
+		Transport: &trailerFixingTransport{next: transport},
 		ModifyResponse: func(res *http.Response) error {
 			addCorsHeaders(res.Header)
 			return nil
@@ -329,11 +331,42 @@ func sendDirectResponse(w http.ResponseWriter, r *http.Request,
 	// so we can use that.
 	switch {
 	case strings.HasPrefix(r.Header.Get(hdrContentType), hdrTypeGrpc):
-		w.Header().Set("Grpc-Status", strconv.Itoa(int(codes.Internal)))
+		w.Header().Set(hdrGrpcStatus, strconv.Itoa(int(codes.Internal)))
-		w.Header().Set("Grpc-Message", errInfo)
+		w.Header().Set(hdrGrpcMessage, errInfo)
+		w.Header().Set("Content-Length", "0")
+		w.Header().Set(":status", strconv.Itoa(statusCode))
+		w.Header().Add("Trailer", hdrGrpcStatus)
+		w.Header().Add("Trailer", hdrGrpcMessage)
+
 		w.WriteHeader(statusCode)
 
 	default:
 		http.Error(w, errInfo, statusCode)
 	}
 }
+
+type trailerFixingTransport struct {
+	next http.RoundTripper
+}
+
+// RoundTrip is a transport round tripper implementation that fixes an issue
+// in the official httputil.ReverseProxy implementation. Apparently the HTTP/2
+// trailers aren't properly forwarded in some cases. We fix this by always
+// copying the Grpc-Status and Grpc-Message fields to the trailers, as those are
+// usually expected to be in the trailer fields.
+// Inspired by https://github.com/elazarl/goproxy/issues/408.
+func (l *trailerFixingTransport) RoundTrip(req *http.Request) (*http.Response,
+	error) {
+
+	resp, err := l.next.RoundTrip(req)
+	if resp != nil && len(resp.Trailer) == 0 {
+		if len(resp.Header.Values(hdrGrpcStatus)) > 0 {
+			resp.Trailer = make(http.Header)
+			grpcStatus := resp.Header.Get(hdrGrpcStatus)
+			grpcMessage := resp.Header.Get(hdrGrpcMessage)
+			resp.Trailer.Add(hdrGrpcStatus, grpcStatus)
+			resp.Trailer.Add(hdrGrpcMessage, grpcMessage)
+		}
+	}
+	return resp, err
+}

proxy/proxy_test.go

@@ -10,6 +10,7 @@ import (
 	"net"
 	"net/http"
 	"path"
+	"strings"
 	"testing"
 	"time"
 
@@ -35,10 +36,15 @@ const (
 	testHTTPResponseBody     = "HTTP Hello"
 )
 
+var (
+	errBackend = fmt.Errorf("this is the error you wanted")
+)
+
 type testCase struct {
-	name          string
+	name           string
-	auth          auth.Level
+	auth           auth.Level
-	authWhitelist []string
+	authWhitelist  []string
+	wantBackendErr bool
 }
 
 // helloServer is a simple server that implements the GreeterServer interface.
@@ -49,6 +55,10 @@ type helloServer struct{}
 func (s *helloServer) SayHello(_ context.Context,
 	req *proxytest.HelloRequest) (*proxytest.HelloReply, error) {
 
+	if req.ReturnError {
+		return nil, errBackend
+	}
+
 	return &proxytest.HelloReply{
 		Message: fmt.Sprintf("Hello %s", req.Name),
 	}, nil
@@ -60,6 +70,10 @@ func (s *helloServer) SayHello(_ context.Context,
 func (s *helloServer) SayHelloNoAuth(_ context.Context,
 	req *proxytest.HelloRequest) (*proxytest.HelloReply, error) {
 
+	if req.ReturnError {
+		return nil, errBackend
+	}
+
 	return &proxytest.HelloReply{
 		Message: fmt.Sprintf("Hello %s", req.Name),
 	}, nil
@@ -180,6 +194,10 @@ func TestProxyGRPC(t *testing.T) {
 	testCases := []*testCase{{
 		name: "no whitelist",
 		auth: "on",
+	}, {
+		name:           "no whitelist expect err",
+		auth:           "on",
+		wantBackendErr: true,
 	}, {
 		name: "with whitelist",
 		auth: "on",
@@ -195,6 +213,17 @@ func TestProxyGRPC(t *testing.T) {
 			runGRPCTest(t, tc)
 		})
 	}
+
+	for i := 0; i < 20; i++ {
+		name := fmt.Sprintf("stream closed w/o trailers repro %d", i)
+		t.Run(name, func(t *testing.T) {
+			runGRPCTest(t, &testCase{
+				name:           name,
+				auth:           "on",
+				wantBackendErr: true,
+			})
+		})
+	}
 }
 
 // TestProxyHTTP tests that the proxy can forward gRPC requests to a backend
@@ -208,7 +237,18 @@ func runGRPCTest(t *testing.T, tc *testCase) {
 	keyFile := path.Join(tempDirName, "proxy.key")
 	certPool, creds, certData, err := genCertPair(certFile, keyFile)
 	require.NoError(t, err)
-	opts := []grpc.DialOption{grpc.WithTransportCredentials(creds)}
+	opts := []grpc.DialOption{
+		grpc.WithTransportCredentials(creds),
+	}
+
+	httpListener, err := net.Listen("tcp", testProxyAddr)
+	if err != nil {
+		t.Errorf("Error listening on %s: %v", testProxyAddr, err)
+	}
+	tlsListener := tls.NewListener(
+		httpListener, configFromCert(&certData, certPool),
+	)
+	defer closeOrFail(t, tlsListener)
 
 	// Create a list of services to proxy between.
 	services := []*proxy.Service{{
@@ -226,20 +266,22 @@ func runGRPCTest(t *testing.T, tc *testCase) {
 	p, err := proxy.New(mockAuth, services, true, "static")
 	require.NoError(t, err)
 	server := &http.Server{
-		Addr:    testProxyAddr,
+		Addr:      testProxyAddr,
-		Handler: http.HandlerFunc(p.ServeHTTP),
+		Handler:   http.HandlerFunc(p.ServeHTTP),
-		TLSConfig: &tls.Config{
+		TLSConfig: configFromCert(&certData, certPool),
-			RootCAs:            certPool,
-			InsecureSkipVerify: true,
-		},
 	}
-	go func() { _ = server.ListenAndServeTLS(certFile, keyFile) }()
+	go func() {
-	defer closeOrFail(t, server)
+		err := server.Serve(tlsListener)
+		if !isClosedErr(err) {
+			t.Errorf("Error serving on %s: %v", testProxyAddr, err)
+		}
+	}()
 
 	// Start the target backend service also on TLS.
-	tlsConf := cert.TLSConfFromCert(certData)
 	serverOpts := []grpc.ServerOption{
-		grpc.Creds(credentials.NewTLS(tlsConf)),
+		grpc.Creds(credentials.NewTLS(configFromCert(
+			&certData, certPool,
+		))),
 	}
 	backendService := grpc.NewServer(serverOpts...)
 	go func() { _ = startBackendGRPC(backendService) }()
@@ -259,8 +301,8 @@ func runGRPCTest(t *testing.T, tc *testCase) {
 	require.Error(t, err)
 	statusErr, ok := status.FromError(err)
 	require.True(t, ok)
-	require.Equal(t, codes.Internal, statusErr.Code())
 	require.Equal(t, "payment required", statusErr.Message())
+	require.Equal(t, codes.Internal, statusErr.Code())
 
 	// Make sure that if we query an URL that is on the whitelist, we don't
 	// get the 402 response.
@@ -292,10 +334,21 @@ func runGRPCTest(t *testing.T, tc *testCase) {
 	client = proxytest.NewGreeterClient(conn)
 
 	// Make the request. This time no error should be returned.
-	req = &proxytest.HelloRequest{Name: "foo"}
+	req = &proxytest.HelloRequest{
+		Name: "foo", ReturnError: tc.wantBackendErr,
+	}
 	res, err := client.SayHello(context.Background(), req)
-	require.NoError(t, err)
+
-	require.Equal(t, "Hello foo", res.Message)
+	if tc.wantBackendErr {
+		require.Error(t, err)
+		statusErr, ok := status.FromError(err)
+		require.True(t, ok)
+		require.Equal(t, errBackend.Error(), statusErr.Message())
+		require.Equal(t, codes.Unknown, statusErr.Code())
+	} else {
+		require.NoError(t, err)
+		require.Equal(t, "Hello foo", res.Message)
+	}
 }
 
 // startBackendHTTP starts the given HTTP server and blocks until the server
@@ -333,7 +386,7 @@ func genCertPair(certFile, keyFile string) (*x509.CertPool,
 	crt := tls.Certificate{}
 	err := cert.GenCertPair(
 		"aperture autogenerated cert", certFile, keyFile, nil, nil,
-		cert.DefaultAutogenValidity,
+		false, cert.DefaultAutogenValidity,
 	)
 	if err != nil {
 		return nil, nil, crt, fmt.Errorf("unable to generate cert "+
@@ -356,9 +409,54 @@ func genCertPair(certFile, keyFile string) (*x509.CertPool,
 	return cp, creds, crt, nil
 }
 
+// configFromCert creates a new TLS configuration from a certificate and a cert
+// pool. These configs shouldn't be shared among different server instances to
+// avoid data races.
+func configFromCert(crt *tls.Certificate, certPool *x509.CertPool) *tls.Config {
+	tlsConf := cert.TLSConfFromCert(*crt)
+	tlsConf.CipherSuites = []uint16{
+		tls.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,
+	}
+	tlsConf.PreferServerCipherSuites = true
+	tlsConf.RootCAs = certPool
+	tlsConf.InsecureSkipVerify = true
+
+	haveNPN := false
+	for _, p := range tlsConf.NextProtos {
+		if p == "h2" {
+			haveNPN = true
+			break
+		}
+	}
+	if !haveNPN {
+		tlsConf.NextProtos = append(tlsConf.NextProtos, "h2")
+	}
+	tlsConf.NextProtos = append(tlsConf.NextProtos, "h2-14")
+	// make sure http 1.1 is *after* all of the other ones.
+	tlsConf.NextProtos = append(tlsConf.NextProtos, "http/1.1")
+
+	return tlsConf
+}
+
 func closeOrFail(t *testing.T, c io.Closer) {
 	err := c.Close()
-	if err != nil {
+	if !isClosedErr(err) {
 		t.Fatal(err)
 	}
 }
+
+func isClosedErr(err error) bool {
+	if err == nil {
+		return true
+	}
+
+	if err == http.ErrServerClosed {
+		return true
+	}
+
+	if strings.Contains(err.Error(), "use of closed network connection") {
+		return true
+	}
+
+	return false
+}

proxy/testdata/gen_protos.sh

@@ -3,6 +3,5 @@
 set -e
 
 protoc -I/usr/local/include -I. \
-       -I$GOPATH/src/github.com/grpc-ecosystem/grpc-gateway/third_party/googleapis \
        --go_out=plugins=grpc,paths=source_relative:. \
        hello.proto

proxy/testdata/hello.pb.go

@@ -26,6 +26,7 @@ const _ = proto.ProtoPackageIsVersion3 // please upgrade the proto package
 
 type HelloRequest struct {
 	Name                 string   `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
+	ReturnError          bool     `protobuf:"varint,2,opt,name=return_error,json=returnError,proto3" json:"return_error,omitempty"`
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
 	XXX_unrecognized     []byte   `json:"-"`
 	XXX_sizecache        int32    `json:"-"`
@@ -63,6 +64,13 @@ func (m *HelloRequest) GetName() string {
 	return ""
 }
 
+func (m *HelloRequest) GetReturnError() bool {
+	if m != nil {
+		return m.ReturnError
+	}
+	return false
+}
+
 type HelloReply struct {
 	Message              string   `protobuf:"bytes,1,opt,name=message,proto3" json:"message,omitempty"`
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
@@ -110,18 +118,19 @@ func init() {
 func init() { proto.RegisterFile("hello.proto", fileDescriptor_61ef911816e0a8ce) }
 
 var fileDescriptor_61ef911816e0a8ce = []byte{
-	// 161 bytes of a gzipped FileDescriptorProto
+	// 186 bytes of a gzipped FileDescriptorProto
 	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe2, 0xe2, 0xce, 0x48, 0xcd, 0xc9,
 	0xc9, 0xd7, 0x2b, 0x28, 0xca, 0x2f, 0xc9, 0x17, 0xe2, 0x2a, 0x28, 0xca, 0xaf, 0xa8, 0x8c, 0x2f,
-	0x49, 0x2d, 0x2e, 0x51, 0x52, 0xe2, 0xe2, 0xf1, 0x00, 0x49, 0x05, 0xa5, 0x16, 0x96, 0xa6, 0x16,
+	0x49, 0x2d, 0x2e, 0x51, 0x72, 0xe5, 0xe2, 0xf1, 0x00, 0x49, 0x05, 0xa5, 0x16, 0x96, 0xa6, 0x16,
 	0x97, 0x08, 0x09, 0x71, 0xb1, 0xe4, 0x25, 0xe6, 0xa6, 0x4a, 0x30, 0x2a, 0x30, 0x6a, 0x70, 0x06,
-	0x81, 0xd9, 0x4a, 0x6a, 0x5c, 0x5c, 0x50, 0x35, 0x05, 0x39, 0x95, 0x42, 0x12, 0x5c, 0xec, 0xb9,
+	0x81, 0xd9, 0x42, 0x8a, 0x5c, 0x3c, 0x45, 0xa9, 0x25, 0xa5, 0x45, 0x79, 0xf1, 0xa9, 0x45, 0x45,
-	0xa9, 0xc5, 0xc5, 0x89, 0xe9, 0x30, 0x45, 0x30, 0xae, 0x51, 0x3f, 0x23, 0x17, 0xbb, 0x7b, 0x51,
+	0xf9, 0x45, 0x12, 0x4c, 0x0a, 0x8c, 0x1a, 0x1c, 0x41, 0xdc, 0x10, 0x31, 0x57, 0x90, 0x90, 0x92,
-	0x6a, 0x6a, 0x49, 0x6a, 0x91, 0x90, 0x1d, 0x17, 0x47, 0x70, 0x62, 0x25, 0x58, 0x9b, 0x90, 0x84,
+	0x1a, 0x17, 0x17, 0xd4, 0x98, 0x82, 0x9c, 0x4a, 0x21, 0x09, 0x2e, 0xf6, 0xdc, 0xd4, 0xe2, 0xe2,
-	0x1e, 0xc2, 0x42, 0x3d, 0x64, 0xdb, 0xa4, 0xc4, 0xb0, 0xc8, 0x14, 0xe4, 0x54, 0x2a, 0x31, 0x08,
+	0xc4, 0x74, 0x98, 0x39, 0x30, 0xae, 0x51, 0x37, 0x23, 0x17, 0xbb, 0x7b, 0x51, 0x6a, 0x6a, 0x49,
-	0xb9, 0x70, 0xf1, 0xc1, 0xf4, 0xfb, 0xe5, 0x3b, 0x96, 0x96, 0x64, 0x90, 0x63, 0x4a, 0x12, 0x1b,
+	0x6a, 0x91, 0x90, 0x0d, 0x17, 0x47, 0x70, 0x62, 0x25, 0x58, 0x9b, 0x90, 0x84, 0x1e, 0xc2, 0x4d,
-	0xd8, 0xc3, 0xc6, 0x80, 0x00, 0x00, 0x00, 0xff, 0xff, 0x61, 0x16, 0xd9, 0xde, 0xff, 0x00, 0x00,
+	0x7a, 0xc8, 0x0e, 0x92, 0x12, 0xc3, 0x22, 0x03, 0xb2, 0xc3, 0x89, 0x8b, 0x0f, 0xa6, 0xdb, 0x2f,
-	0x00,
+	0xdf, 0xb1, 0xb4, 0x24, 0x83, 0x74, 0x33, 0x92, 0xd8, 0xc0, 0xe1, 0x61, 0x0c, 0x08, 0x00, 0x00,
+	0xff, 0xff, 0x83, 0xcb, 0x43, 0x7d, 0x1e, 0x01, 0x00, 0x00,
 }
 
 // Reference imports to suppress errors if they are not otherwise used.

proxy/testdata/hello.proto

@@ -3,12 +3,13 @@ syntax = "proto3";
 package proxy_test;
 
 service Greeter {
-    rpc SayHello (HelloRequest) returns (HelloReply) {}
+    rpc SayHello (HelloRequest) returns (HelloReply);
-    rpc SayHelloNoAuth (HelloRequest) returns (HelloReply) {}
+    rpc SayHelloNoAuth (HelloRequest) returns (HelloReply);
 }
 
 message HelloRequest {
     string name = 1;
+    bool return_error = 2;
 }
 
 message HelloReply {