Make maximum gRPC message size configurable (TykTechnologies#2204)

Potential fix for TykTechnologies#2203 

This introduces two new parameters in `tyk.conf`:
```json
{
  "grpc_recv_max_size": 100000,
  "grpc_send_max_size": 100000
}
```
When no parameter is set (and both parameters equal `0`), we call `WithDefaultCallOptions` so default values are used.
The value is a number of bytes.

cli/linter/schema.json

@@ -257,6 +257,12 @@
         },
         "python_version": {
           "type": "string"
+        },
+        "grpc_recv_max_size": {
+          "type": "integer"
+        },
+        "grpc_send_max_size": {
+          "type": "integer"
         }
       }
     },

config/config.go

@@ -214,6 +214,8 @@ type ServiceDiscoveryConf struct {
 type CoProcessConfig struct {
 	EnableCoProcess     bool   `json:"enable_coprocess"`
 	CoProcessGRPCServer string `json:"coprocess_grpc_server"`
+	GRPCRecvMaxSize     int    `json:"grpc_recv_max_size"`
+	GRPCSendMaxSize     int    `json:"grpc_send_max_size"`
 	PythonPathPrefix    string `json:"python_path_prefix"`
 	PythonVersion       string `json:"python_version"`
 }

coprocess/grpc/coprocess_grpc_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"encoding/json"
 	"io/ioutil"
+	"math/rand"
 	"mime/multipart"
 	"net"
 	"net/http"
@@ -24,8 +25,9 @@ import (
 )
 
 const (
-	grpcListenAddr = ":9999"
-	grpcListenPath = "tcp://127.0.0.1:9999"
+	grpcListenAddr  = ":9999"
+	grpcListenPath  = "tcp://127.0.0.1:9999"
+	grpcTestMaxSize = 100000000
 
 	testHeaderName  = "Testheader"
 	testHeaderValue = "testvalue"
@@ -115,7 +117,10 @@ func (d *dispatcher) DispatchEvent(ctx context.Context, event *coprocess.Event)
 }
 
 func newTestGRPCServer() (s *grpc.Server) {
-	s = grpc.NewServer()
+	s = grpc.NewServer(
+		grpc.MaxRecvMsgSize(grpcTestMaxSize),
+		grpc.MaxSendMsgSize(grpcTestMaxSize),
+	)
 	coprocess.RegisterDispatcherServer(s, &dispatcher{})
 	return s
 }
@@ -244,6 +249,8 @@ func startTykWithGRPC() (*gateway.Test, *grpc.Server) {
 	cfg := config.CoProcessConfig{
 		EnableCoProcess:     true,
 		CoProcessGRPCServer: grpcListenPath,
+		GRPCRecvMaxSize:     grpcTestMaxSize,
+		GRPCSendMaxSize:     grpcTestMaxSize,
 	}
 	ts := gateway.StartTest(gateway.TestConfig{CoprocessConfig: cfg})
 
@@ -347,6 +354,27 @@ func TestGRPCDispatch(t *testing.T) {
 		})
 	})
 
+	t.Run("Post Hook with allowed message length", func(t *testing.T) {
+		s := randStringBytes(20000000)
+		ts.Run(t, test.TestCase{
+			Path:    "/grpc-test-api-3/",
+			Method:  http.MethodGet,
+			Code:    http.StatusOK,
+			Headers: headers,
+			Data:    s,
+		})
+	})
+
+	t.Run("Post Hook with with unallowed message length", func(t *testing.T) {
+		s := randStringBytes(150000000)
+		ts.Run(t, test.TestCase{
+			Path:    "/grpc-test-api-3/",
+			Method:  http.MethodGet,
+			Code:    http.StatusInternalServerError,
+			Headers: headers,
+			Data:    s,
+		})
+	})
 }
 
 func BenchmarkGRPCDispatch(b *testing.B) {
@@ -370,3 +398,15 @@ func BenchmarkGRPCDispatch(b *testing.B) {
 		}
 	})
 }
+
+const letters = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
+
+func randStringBytes(n int) string {
+	b := make([]byte, n)
+
+	for i := range b {
+		b[i] = letters[rand.Intn(len(letters))]
+	}
+
+	return string(b)
+}

gateway/coprocess_grpc.go

@@ -72,13 +72,30 @@ func (d *GRPCDispatcher) Reload() {}
 // HandleMiddlewareCache isn't used by gRPC.
 func (d *GRPCDispatcher) HandleMiddlewareCache(b *apidef.BundleManifest, basePath string) {}
 
+func grpcCallOpts() grpc.DialOption {
+	recvSize := config.Global().CoProcessOptions.GRPCRecvMaxSize
+	sendSize := config.Global().CoProcessOptions.GRPCSendMaxSize
+	var opts []grpc.CallOption
+	if recvSize > 0 {
+		opts = append(opts, grpc.MaxCallRecvMsgSize(recvSize))
+	}
+	if sendSize > 0 {
+		opts = append(opts, grpc.MaxCallRecvMsgSize(sendSize))
+	}
+	return grpc.WithDefaultCallOptions(opts...)
+}
+
 // NewGRPCDispatcher wraps all the actions needed for this CP.
 func NewGRPCDispatcher() (coprocess.Dispatcher, error) {
 	if config.Global().CoProcessOptions.CoProcessGRPCServer == "" {
 		return nil, errors.New("No gRPC URL is set")
 	}
 	var err error
-	grpcConnection, err = grpc.Dial("", grpc.WithInsecure(), grpc.WithDialer(dialer))
+	grpcConnection, err = grpc.Dial("",
+		grpcCallOpts(),
+		grpc.WithInsecure(),
+		grpc.WithDialer(dialer),
+	)
 	grpcClient = coprocess.NewDispatcherClient(grpcConnection)
 
 	if err != nil {

gateway/mw_strip_auth_test.go

@@ -2,7 +2,6 @@ package gateway
 
 import (
 	"fmt"
-	"math/rand"
 	"net/http"
 	"net/url"
 	"testing"
@@ -16,18 +15,6 @@ type TestAuth struct {
 	QueryParam string
 }
 
-const letters = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
-
-func randStringBytes(n int) string {
-	b := make([]byte, n)
-
-	for i := range b {
-		b[i] = letters[rand.Intn(len(letters))]
-	}
-
-	return string(b)
-}
-
 func testPrepareStripAuthStripFromHeaders() ([]string, []TestAuth) {
 	testCases := []TestAuth{
 		{AuthConfig: apidef.AuthConfig{AuthHeaderName: "Authorization"}, HeaderKey: "Authorization"},

gateway/testutil.go

@@ -982,3 +982,14 @@ YGivtXBGXk1hlVYlje1RB+W6RQuDAegI5h8vl8pYJS9JQH0wjatsDaE=
 `
 
 const jwtSecret = "9879879878787878"
+const letters = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
+
+func randStringBytes(n int) string {
+	b := make([]byte, n)
+
+	for i := range b {
+		b[i] = letters[rand.Intn(len(letters))]
+	}
+
+	return string(b)
+}