e2e: add gRPC tests

The tests are ported from Martian h2 tests.
The client is called fixture for compatibility reasons.

e2e/certs/gen.sh

@@ -56,5 +56,6 @@ EOF
 generate_certificate "proxy"
 generate_certificate "upstream-proxy"
 generate_certificate "httpbin"
+generate_certificate "grpctest"
 
 chmod 644 *.key *.crt

e2e/forwarder/service.go

@@ -21,6 +21,7 @@ const (
 	ProxyServiceName         = "proxy"
 	UpstreamProxyServiceName = "upstream-proxy"
 	HttpbinServiceName       = "httpbin"
+	GRPCTestServiceName      = "grpctest"
 )
 
 const enabled = "true"
@@ -70,6 +71,22 @@ func HttpbinService() *Service {
 	}
 }
 
+func GRPCTestService() *Service {
+	s := &Service{
+		Name:    GRPCTestServiceName,
+		Image:   Image,
+		Command: "test grpc",
+		Environment: map[string]string{
+			"FORWARDER_ADDRESS": ":1443",
+		},
+		Ports: []string{
+			"1443:1443",
+			"10003:10000",
+		},
+	}
+	return s.WithProtocol("h2")
+}
+
 func (s *Service) WithProtocol(protocol string) *Service {
 	s.Environment["FORWARDER_PROTOCOL"] = protocol
 

e2e/setups.go

@@ -33,6 +33,7 @@ func AllSetups() []setup.Setup {
 	SetupDefaults(l)
 	SetupAuth(l)
 	SetupPac(l)
+	SetupGRPC(l)
 	SetupFlagProxyLocalhost(l)
 	SetupFlagHeader(l)
 	SetupFlagResponseHeader(l)
@@ -164,6 +165,19 @@ func SetupPac(l *setupList) {
 	)
 }
 
+func SetupGRPC(l *setupList) {
+	l.Add(
+		setup.Setup{
+			Name: "grpc",
+			Compose: compose.NewBuilder().
+				AddService(forwarder.ProxyService()).
+				AddService(forwarder.GRPCTestService()).
+				MustBuild(),
+			Run: "^TestGRPC",
+		},
+	)
+}
+
 func SetupFlagProxyLocalhost(l *setupList) {
 	for _, mode := range []string{"deny", "allow"} {
 		l.Add(setup.Setup{

e2e/tests/grpc_test.go

@@ -0,0 +1,144 @@
+// Copyright 2023 Sauce Labs Inc., all rights reserved.
+//
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at https://mozilla.org/MPL/2.0/.
+
+package tests
+
+import (
+	"context"
+	"crypto/rand"
+	"crypto/tls"
+	"encoding/base64"
+	"errors"
+	"io"
+	"sync"
+	"testing"
+
+	"github.com/saucelabs/forwarder/e2e/forwarder"
+	tspb "github.com/saucelabs/forwarder/internal/martian/h2/testservice"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/encoding/gzip"
+)
+
+func TestGRPC(t *testing.T) {
+	tlsCfg := &tls.Config{
+		InsecureSkipVerify: true,
+	}
+
+	t.Setenv("HTTPS_PROXY", proxy) // set proxy for grpc.Dial
+	conn, err := grpc.Dial(forwarder.GRPCTestServiceName+":1443", grpc.WithTransportCredentials(credentials.NewTLS(tlsCfg)))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	fixture := tspb.NewTestServiceClient(conn)
+
+	t.Run("Echo", func(t *testing.T) {
+		ctx := context.Background()
+		req := &tspb.EchoRequest{
+			Payload: "Hello",
+		}
+		resp, err := fixture.Echo(ctx, req)
+		if err != nil {
+			t.Fatalf("fixture.Echo(...) = _, %v, want _, nil", err)
+		}
+		if got, want := resp.GetPayload(), req.GetPayload(); got != want {
+			t.Errorf("resp.GetPayload() = %s, want = %s", got, want)
+		}
+	})
+
+	t.Run("LargeEcho", func(t *testing.T) {
+		// Sends a >128KB payload through the proxy. Since the standard gRPC frame size is only 16KB,
+		// this exercises frame merging, splitting and flow control code.
+		payload := make([]byte, 128*1024)
+		rand.Read(payload)
+		req := &tspb.EchoRequest{
+			Payload: base64.StdEncoding.EncodeToString(payload),
+		}
+
+		// This test also covers using gzip compression. Ideally, we would test more compression types
+		// but the golang gRPC implementation only provides a gzip compressor.
+		tests := []struct {
+			name           string
+			useCompression bool
+		}{
+			{"RawData", false},
+			{"Gzip", true},
+		}
+		for _, tc := range tests {
+			t.Run(tc.name, func(t *testing.T) {
+				ctx := context.Background()
+				var resp *tspb.EchoResponse
+				if tc.useCompression {
+					resp, err = fixture.Echo(ctx, req, grpc.UseCompressor(gzip.Name))
+				} else {
+					resp, err = fixture.Echo(ctx, req)
+				}
+				if err != nil {
+					t.Fatalf("fixture.Echo(...) = _, %v, want _, nil", err)
+				}
+				if got, want := resp.GetPayload(), req.GetPayload(); got != want {
+					t.Errorf("resp.GetPayload() = %s, want = %s", got, want)
+				}
+			})
+		}
+	})
+
+	t.Run("Stream", func(t *testing.T) {
+		ctx := context.Background()
+		stream, err := fixture.DoubleEcho(ctx)
+		if err != nil {
+			t.Fatalf("fixture.DoubleEcho(ctx) = _, %v, want _, nil", err)
+		}
+
+		var received []*tspb.EchoResponse
+
+		var wg sync.WaitGroup
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for {
+				resp, err := stream.Recv()
+				if errors.Is(err, io.EOF) {
+					return
+				}
+				if err != nil {
+					t.Errorf("stream.Recv() = %v, want nil", err)
+					return
+				}
+				received = append(received, resp)
+			}
+		}()
+
+		var sent []*tspb.EchoRequest
+		for i := 0; i < 5; i++ {
+			payload := make([]byte, 20*1024)
+			rand.Read(payload)
+			req := &tspb.EchoRequest{
+				Payload: base64.StdEncoding.EncodeToString(payload),
+			}
+			if err := stream.Send(req); err != nil {
+				t.Fatalf("stream.Send(req) = %v, want nil", err)
+			}
+			sent = append(sent, req)
+		}
+		if err := stream.CloseSend(); err != nil {
+			t.Fatalf("stream.CloseSend() = %v, want nil", err)
+		}
+		wg.Wait()
+
+		for i, req := range sent {
+			want := req.GetPayload()
+			if got := received[2*i].GetPayload(); got != want {
+				t.Errorf("received[2*i].GetPayload() = %s, want %s", got, want)
+			}
+			if got := received[2*i+1].GetPayload(); got != want {
+				t.Errorf("received[2*i+1].GetPayload() = %s, want %s", got, want)
+			}
+		}
+	})
+}