Multiplexing opt out flag (#16972)

* added mplexing opt-out flag

changelog/16972.txt

@@ -0,0 +1,3 @@
+```release-note:improvement
+plugins: Added environment variable flag to opt-out specific plugins from multiplexing
+```

go.mod

@@ -424,4 +424,4 @@ require (
 	k8s.io/klog/v2 v2.60.1 // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.2.1 // indirect
 	sigs.k8s.io/yaml v1.2.0 // indirect
-)
+)

go.sum

@@ -2663,4 +2663,4 @@ sigs.k8s.io/structured-merge-diff/v4 v4.2.1 h1:bKCqE9GvQ5tiVHn5rfn1r+yao3aLQEaLz
 sigs.k8s.io/structured-merge-diff/v4 v4.2.1/go.mod h1:j/nl6xW8vLS49O8YvXW1ocPhZawJtm+Yrr7PPRQ0Vg4=
 sigs.k8s.io/yaml v1.1.0/go.mod h1:UJmg0vDUVViEyp3mgSv9WPwZCDxu4rQW1olrI1uml+o=
 sigs.k8s.io/yaml v1.2.0 h1:kr/MCeFWJWTwyaHoR9c8EjH9OumOmoF9YGiZd7lFm/Q=
-sigs.k8s.io/yaml v1.2.0/go.mod h1:yfXDCHCao9+ENCvLSE62v9VSji2MKu5jeNfTrofGhJc=
+sigs.k8s.io/yaml v1.2.0/go.mod h1:yfXDCHCao9+ENCvLSE62v9VSji2MKu5jeNfTrofGhJc=

sdk/go.mod

@@ -1,6 +1,6 @@
 module github.com/hashicorp/vault/sdk
 
-go 1.17
+go 1.19
 
 require (
 	github.com/armon/go-metrics v0.3.9

sdk/go.sum

@@ -317,4 +317,4 @@ gopkg.in/yaml.v2 v2.2.5/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
-honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
+honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=

sdk/helper/pluginutil/env.go

@@ -31,6 +31,10 @@ const (
 	// PluginCACertPEMEnv is an ENV name used for holding a CA PEM-encoded
 	// string. Used for testing.
 	PluginCACertPEMEnv = "VAULT_TESTING_PLUGIN_CA_PEM"
+
+	// PluginMultiplexingOptOut is an ENV name used to define a comma separated list of plugin names
+	// opted-out of the multiplexing feature; for emergencies if multiplexing ever causes issues
+	PluginMultiplexingOptOut = "VAULT_PLUGIN_MULTIPLEXING_OPT_OUT"
 )
 
 // OptionallyEnableMlock determines if mlock should be called, and if so enables

sdk/helper/pluginutil/multiplexing.go

@@ -1,13 +1,16 @@
 package pluginutil
 
 import (
-	context "context"
+	"context"
 	"fmt"
+	"os"
+	"strings"
 
-	grpc "google.golang.org/grpc"
+	"github.com/hashicorp/go-secure-stdlib/strutil"
-	codes "google.golang.org/grpc/codes"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/metadata"
-	status "google.golang.org/grpc/status"
+	"google.golang.org/grpc/status"
 )
 
 type PluginMultiplexingServerImpl struct {
@@ -16,17 +19,22 @@ type PluginMultiplexingServerImpl struct {
 	Supported bool
 }
 
-func (pm PluginMultiplexingServerImpl) MultiplexingSupport(ctx context.Context, req *MultiplexingSupportRequest) (*MultiplexingSupportResponse, error) {
+func (pm PluginMultiplexingServerImpl) MultiplexingSupport(_ context.Context, _ *MultiplexingSupportRequest) (*MultiplexingSupportResponse, error) {
 	return &MultiplexingSupportResponse{
 		Supported: pm.Supported,
 	}, nil
 }
 
-func MultiplexingSupported(ctx context.Context, cc grpc.ClientConnInterface) (bool, error) {
+func MultiplexingSupported(ctx context.Context, cc grpc.ClientConnInterface, name string) (bool, error) {
 	if cc == nil {
 		return false, fmt.Errorf("client connection is nil")
 	}
 
+	out := strings.Split(os.Getenv(PluginMultiplexingOptOut), ",")
+	if strutil.StrListContains(out, name) {
+		return false, nil
+	}
+
 	req := new(MultiplexingSupportRequest)
 	resp, err := NewPluginMultiplexingClient(cc).MultiplexingSupport(ctx, req)
 	if err != nil {

sdk/helper/pluginutil/multiplexing_test.go

@@ -6,9 +6,82 @@ import (
 	"reflect"
 	"testing"
 
+	"google.golang.org/grpc"
 	"google.golang.org/grpc/metadata"
 )
 
+func TestMultiplexingSupported(t *testing.T) {
+	type args struct {
+		ctx  context.Context
+		cc   grpc.ClientConnInterface
+		name string
+	}
+
+	type testCase struct {
+		name    string
+		args    args
+		env     string
+		want    bool
+		wantErr bool
+	}
+
+	tests := []testCase{
+		{
+			name: "multiplexing is supported if plugin is not opted out",
+			args: args{
+				ctx:  context.Background(),
+				cc:   &MockClientConnInterfaceNoop{},
+				name: "plugin",
+			},
+			env:  "",
+			want: true,
+		},
+		{
+			name: "multiplexing is not supported if plugin is opted out",
+			args: args{
+				ctx:  context.Background(),
+				cc:   &MockClientConnInterfaceNoop{},
+				name: "optedOutPlugin",
+			},
+			env:  "optedOutPlugin",
+			want: false,
+		},
+		{
+			name: "multiplexing is not supported if plugin among one of the opted out",
+			args: args{
+				ctx:  context.Background(),
+				cc:   &MockClientConnInterfaceNoop{},
+				name: "optedOutPlugin",
+			},
+			env:  "firstPlugin,optedOutPlugin,otherPlugin",
+			want: false,
+		},
+		{
+			name: "multiplexing is supported if different plugin is opted out",
+			args: args{
+				ctx:  context.Background(),
+				cc:   &MockClientConnInterfaceNoop{},
+				name: "plugin",
+			},
+			env:  "optedOutPlugin",
+			want: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Setenv(PluginMultiplexingOptOut, tt.env)
+			got, err := MultiplexingSupported(tt.args.ctx, tt.args.cc, tt.args.name)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("MultiplexingSupported() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if got != tt.want {
+				t.Errorf("MultiplexingSupported() got = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
 func TestGetMultiplexIDFromContext(t *testing.T) {
 	type testCase struct {
 		ctx          context.Context
@@ -71,3 +144,14 @@ func idCtx(t *testing.T, ids ...string) context.Context {
 	}
 	return metadata.NewIncomingContext(ctx, md)
 }
+
+type MockClientConnInterfaceNoop struct{}
+
+func (m *MockClientConnInterfaceNoop) Invoke(_ context.Context, _ string, _ interface{}, reply interface{}, _ ...grpc.CallOption) error {
+	reply.(*MultiplexingSupportResponse).Supported = true
+	return nil
+}
+
+func (m *MockClientConnInterfaceNoop) NewStream(_ context.Context, _ *grpc.StreamDesc, _ string, _ ...grpc.CallOption) (grpc.ClientStream, error) {
+	return nil, nil
+}

vault/plugin_catalog.go

@@ -348,20 +348,14 @@ func (c *PluginCatalog) newPluginClient(ctx context.Context, pluginRunner *plugi
 
 	clientConn := rpcClient.(*plugin.GRPCClient).Conn
 
-	muxed, err := pluginutil.MultiplexingSupported(ctx, clientConn)
+	muxed, err := pluginutil.MultiplexingSupported(ctx, clientConn, config.Name)
 	if err != nil {
 		return nil, err
 	}
 
-	if muxed {
+	pc.clientConn = &pluginClientConn{
-		// Wrap rpcClient with our implementation so that we can inject the
+		ClientConn: clientConn,
-		// ID into the context
+		id:         id,
-		pc.clientConn = &pluginClientConn{
-			ClientConn: clientConn,
-			id:         id,
-		}
-	} else {
-		pc.clientConn = clientConn
 	}
 
 	pc.ClientProtocol = rpcClient