Add gRPC e2e test

command/forwarder/root.go

@@ -8,10 +8,11 @@ package forwarder
 
 import (
 	"github.com/saucelabs/forwarder/bind"
-	"github.com/saucelabs/forwarder/command/httpbin"
 	"github.com/saucelabs/forwarder/command/pac"
 	"github.com/saucelabs/forwarder/command/ready"
 	"github.com/saucelabs/forwarder/command/run"
+	"github.com/saucelabs/forwarder/command/test/grpc"
+	"github.com/saucelabs/forwarder/command/test/httpbin"
 	"github.com/saucelabs/forwarder/command/version"
 	"github.com/saucelabs/forwarder/utils/cobrautil"
 	"github.com/saucelabs/forwarder/utils/cobrautil/templates"
@@ -105,11 +106,19 @@ func Command() *cobra.Command {
 
 	templates.ActsAsRootCommand(cmd, nil, cg, FlagGroups(), EnvPrefix)
 
-	// Add other commands.
-	cmd.AddCommand(
-		httpbin.Command(), // hidden
-		version.Command(),
+	// Add test commands.
+	test := &cobra.Command{
+		Use:   "test",
+		Short: "Run test servers for various protocols",
+	}
+	test.AddCommand(
+		grpc.Command(),
+		httpbin.Command(),
 	)
+	cmd.AddCommand(test)
+
+	// Add version command.
+	cmd.AddCommand(version.Command())
 
 	// Add config-file command to all commands.
 	cobrautil.AddConfigFileForEachCommand(cmd, FlagGroups(), ConfigFileFlagName)

command/test/grpc/grpc.go

@@ -0,0 +1,124 @@
+// 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 grpc
+
+import (
+	"context"
+	"crypto/tls"
+	"net"
+
+	"github.com/saucelabs/forwarder"
+	"github.com/saucelabs/forwarder/bind"
+	ts "github.com/saucelabs/forwarder/internal/martian/h2/testing"
+	tspb "github.com/saucelabs/forwarder/internal/martian/h2/testservice"
+	"github.com/saucelabs/forwarder/log"
+	"github.com/saucelabs/forwarder/log/stdlog"
+	"github.com/saucelabs/forwarder/runctx"
+	"github.com/saucelabs/forwarder/utils/cobrautil"
+	"github.com/spf13/cobra"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	_ "google.golang.org/grpc/encoding/gzip" // register gzip encoding
+)
+
+type command struct {
+	addr            string
+	tlsServerConfig *forwarder.TLSServerConfig
+	logConfig       *log.Config
+}
+
+func (c *command) runE(cmd *cobra.Command, _ []string) (cmdErr error) {
+	if f := c.logConfig.File; f != nil {
+		defer f.Close()
+	}
+	logger := stdlog.New(c.logConfig)
+
+	defer func() {
+		if cmdErr != nil {
+			logger.Errorf("fatal error exiting: %s", cmdErr)
+			cmd.SilenceErrors = true
+		}
+	}()
+
+	{
+		var (
+			cfg []byte
+			err error
+		)
+
+		d := cobrautil.FlagsDescriber{
+			Format: cobrautil.Plain,
+		}
+		cfg, err = d.DescribeFlags(cmd.Flags())
+		if err != nil {
+			return err
+		}
+		logger.Infof("configuration\n%s", cfg)
+
+		d.ShowNotChanged = true
+		cfg, err = d.DescribeFlags(cmd.Flags())
+		if err != nil {
+			return err
+		}
+		logger.Debugf("all configuration\n%s\n\n", cfg)
+	}
+
+	g := runctx.NewGroup()
+
+	{
+		tlsCfg := new(tls.Config)
+		if err := c.tlsServerConfig.ConfigureTLSConfig(tlsCfg); err != nil {
+			return err
+		}
+		gs := grpc.NewServer(
+			grpc.Creds(credentials.NewServerTLSFromCert(&tlsCfg.Certificates[0])),
+		)
+		tspb.RegisterTestServiceServer(gs, &ts.Server{})
+		defer gs.Stop()
+
+		l, err := net.Listen("tcp", c.addr)
+		if err != nil {
+			return err
+		}
+		defer l.Close()
+
+		g.Add(func(ctx context.Context) error {
+			logger.Named("grpc").Infof("server listen address=%s", l.Addr())
+			go func() {
+				<-ctx.Done()
+				gs.GracefulStop()
+			}()
+			return gs.Serve(l)
+		})
+	}
+
+	return g.Run()
+}
+
+func Command() *cobra.Command {
+	c := command{
+		addr:            "localhost:1443",
+		tlsServerConfig: new(forwarder.TLSServerConfig),
+		logConfig:       log.DefaultConfig(),
+	}
+
+	cmd := &cobra.Command{
+		Use:   "grpc [--address <host:port>] [flags]",
+		Short: "Start HTTP/2 gRPC server for testing",
+		RunE:  c.runE,
+	}
+
+	fs := cmd.Flags()
+	fs.StringVar(&c.addr, "address", c.addr, "<host:port>"+
+		"Address to listen on. "+
+		"If the host is empty, the server will listen on all available interfaces. ")
+	bind.TLSServerConfig(fs, c.tlsServerConfig, "")
+	bind.LogConfig(fs, c.logConfig)
+	bind.AutoMarkFlagFilename(cmd)
+
+	return cmd
+}

command/httpbin/httpbin.go → command/test/httpbin/httpbin.go

@@ -87,10 +87,9 @@ func Command() *cobra.Command {
 	c.apiServerConfig.Addr = "localhost:10000"
 
 	cmd := &cobra.Command{
-		Use:    "httpbin [--protocol <http|https|h2>] [--address <host:port>] [flags]",
-		Short:  "Start HTTP(S) server that serves httpbin.org API",
-		RunE:   c.runE,
-		Hidden: true,
+		Use:   "httpbin [--protocol <http|https|h2>] [--address <host:port>] [flags]",
+		Short: "Start HTTP(S) server that serves httpbin.org API",
+		RunE:  c.runE,
 	}
 
 	fs := cmd.Flags()

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"
@@ -59,7 +60,7 @@ func HttpbinService() *Service {
 	return &Service{
 		Name:    HttpbinServiceName,
 		Image:   Image,
-		Command: "httpbin",
+		Command: "test httpbin",
 		Environment: map[string]string{
 			"FORWARDER_API_ADDRESS": ":10000",
 		},
@@ -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,146 @@
+// 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/.
+
+//go:build e2e
+
+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)
+			}
+		}
+	})
+}