Merge pull request #376 from yacovm/tlsRotate

[FAB-17220] Dynamically build TLS config in Raft client handshake

core/comm/client.go

@@ -14,7 +14,6 @@ import (
 
 	"github.com/pkg/errors"
 	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/keepalive"
 )
 
@@ -188,12 +187,11 @@ func (client *GRPCClient) NewConnection(address string, tlsOptions ...TLSOption)
 	// SetServerRootCAs / SetMaxRecvMsgSize / SetMaxSendMsgSize
 	//  to take effect on a per connection basis
 	if client.tlsConfig != nil {
-		tlsConfigCopy := client.tlsConfig.Clone()
-		for _, tlsOption := range tlsOptions {
-			tlsOption(tlsConfigCopy)
-		}
 		dialOpts = append(dialOpts, grpc.WithTransportCredentials(
-			credentials.NewTLS(tlsConfigCopy),
+			&DynamicClientCredentials{
+				TLSConfig:  client.tlsConfig,
+				TLSOptions: tlsOptions,
+			},
 		))
 	} else {
 		dialOpts = append(dialOpts, grpc.WithInsecure())

core/comm/client_test.go

@@ -14,16 +14,21 @@ import (
 	"io/ioutil"
 	"net"
 	"path/filepath"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/golang/protobuf/proto"
+	"github.com/hyperledger/fabric/common/crypto/tlsgen"
+	"github.com/hyperledger/fabric/common/flogging"
 	"github.com/hyperledger/fabric/core/comm"
 	"github.com/hyperledger/fabric/core/comm/testpb"
 	"github.com/pkg/errors"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials"
 )
 
@@ -579,3 +584,86 @@ func TestCertPoolOverride(t *testing.T) {
 		RootCAs: &x509.CertPool{},
 	}, testConfig)
 }
+
+func TestDynamicClientTLSLoading(t *testing.T) {
+	t.Parallel()
+	ca1, err := tlsgen.NewCA()
+	assert.NoError(t, err)
+
+	ca2, err := tlsgen.NewCA()
+	assert.NoError(t, err)
+
+	clientKP, err := ca1.NewClientCertKeyPair()
+	assert.NoError(t, err)
+
+	serverKP, err := ca2.NewServerCertKeyPair("127.0.0.1")
+	assert.NoError(t, err)
+
+	client, err := comm.NewGRPCClient(comm.ClientConfig{
+		AsyncConnect: true,
+		Timeout:      time.Second * 1,
+		SecOpts: comm.SecureOptions{
+			UseTLS:        true,
+			ServerRootCAs: [][]byte{ca1.CertBytes()},
+			Certificate:   clientKP.Cert,
+			Key:           clientKP.Key,
+		},
+	})
+	assert.NoError(t, err)
+
+	server, err := comm.NewGRPCServer("127.0.0.1:0", comm.ServerConfig{
+		Logger: flogging.MustGetLogger("test"),
+		SecOpts: comm.SecureOptions{
+			UseTLS:      true,
+			Key:         serverKP.Key,
+			Certificate: serverKP.Cert,
+		},
+	})
+	assert.NoError(t, err)
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+
+	go func() {
+		defer wg.Done()
+		server.Start()
+	}()
+
+	var dynamicRootCerts atomic.Value
+	dynamicRootCerts.Store(ca1.CertBytes())
+
+	conn, err := client.NewConnection(server.Address(), func(tlsConfig *tls.Config) {
+		tlsConfig.RootCAs = x509.NewCertPool()
+		tlsConfig.RootCAs.AppendCertsFromPEM(dynamicRootCerts.Load().([]byte))
+	})
+	assert.NoError(t, err)
+	assert.NotNil(t, conn)
+
+	waitForConnState := func(state connectivity.State, succeedOrFail string) {
+		deadline := time.Now().Add(time.Second * 30)
+		for conn.GetState() != state {
+			time.Sleep(time.Millisecond * 10)
+			if time.Now().After(deadline) {
+				t.Fatalf("Test timed out, waited for connection to %s", succeedOrFail)
+			}
+		}
+	}
+
+	// Poll the connection state to wait for it to fail
+	waitForConnState(connectivity.TransientFailure, "fail")
+
+	// Update the TLS root CAs with the good one
+	dynamicRootCerts.Store(ca2.CertBytes())
+
+	// Reset exponential back-off to make the test faster
+	conn.ResetConnectBackoff()
+
+	// Poll the connection state to wait for it to succeed
+	waitForConnState(connectivity.Ready, "succeed")
+
+	err = conn.Close()
+	assert.NoError(t, err)
+
+	server.Stop()
+	wg.Wait()
+}

core/comm/creds.go

@@ -18,6 +18,7 @@ import (
 
 var (
 	ErrClientHandshakeNotImplemented = errors.New("core/comm: client handshakes are not implemented with serverCreds")
+	ErrServerHandshakeNotImplemented = errors.New("core/comm: server handshakes are not implemented with clientCreds")
 	ErrOverrideHostnameNotSupported  = errors.New("core/comm: OverrideServerName is not supported")
 
 	// alpnProtoStr are the specified application level protocols for gRPC.
@@ -85,3 +86,37 @@ func (sc *serverCreds) Clone() credentials.TransportCredentials {
 func (sc *serverCreds) OverrideServerName(string) error {
 	return ErrOverrideHostnameNotSupported
 }
+
+type DynamicClientCredentials struct {
+	TLSConfig  *tls.Config
+	TLSOptions []TLSOption
+}
+
+func (dtc *DynamicClientCredentials) latestConfig() *tls.Config {
+	tlsConfigCopy := dtc.TLSConfig.Clone()
+	for _, tlsOption := range dtc.TLSOptions {
+		tlsOption(tlsConfigCopy)
+	}
+	return tlsConfigCopy
+}
+
+func (dtc *DynamicClientCredentials) ClientHandshake(ctx context.Context, authority string, rawConn net.Conn) (net.Conn, credentials.AuthInfo, error) {
+	return credentials.NewTLS(dtc.latestConfig()).ClientHandshake(ctx, authority, rawConn)
+}
+
+func (dtc *DynamicClientCredentials) ServerHandshake(rawConn net.Conn) (net.Conn, credentials.AuthInfo, error) {
+	return nil, nil, ErrServerHandshakeNotImplemented
+}
+
+func (dtc *DynamicClientCredentials) Info() credentials.ProtocolInfo {
+	return credentials.NewTLS(dtc.latestConfig()).Info()
+}
+
+func (dtc *DynamicClientCredentials) Clone() credentials.TransportCredentials {
+	return credentials.NewTLS(dtc.latestConfig())
+}
+
+func (dtc *DynamicClientCredentials) OverrideServerName(name string) error {
+	dtc.TLSConfig.ServerName = name
+	return nil
+}

integration/nwo/configblock.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/golang/protobuf/proto"
 	"github.com/hyperledger/fabric-protos-go/common"
+	"github.com/hyperledger/fabric-protos-go/msp"
 	protosorderer "github.com/hyperledger/fabric-protos-go/orderer"
 	"github.com/hyperledger/fabric/integration/nwo/commands"
 	"github.com/hyperledger/fabric/internal/configtxlator/update"
@@ -290,6 +291,9 @@ func UnmarshalBlockFromFile(blockFile string) *common.Block {
 // ConsensusMetadataMutator receives ConsensusType.Metadata and mutates it.
 type ConsensusMetadataMutator func([]byte) []byte
 
+// MSPMutator receives FabricMSPConfig and mutates it.
+type MSPMutator func(config msp.FabricMSPConfig) msp.FabricMSPConfig
+
 // UpdateConsensusMetadata executes a config update that updates the consensus
 // metadata according to the given ConsensusMetadataMutator.
 func UpdateConsensusMetadata(network *Network, peer *Peer, orderer *Orderer, channel string, mutateMetadata ConsensusMetadataMutator) {
@@ -310,3 +314,27 @@ func UpdateConsensusMetadata(network *Network, peer *Peer, orderer *Orderer, cha
 
 	UpdateOrdererConfig(network, orderer, channel, config, updatedConfig, peer, orderer)
 }
+
+func UpdateOrdererMSP(network *Network, peer *Peer, orderer *Orderer, channel, orgID string, mutateMSP MSPMutator) {
+	config := GetConfig(network, peer, orderer, channel)
+	updatedConfig := proto.Clone(config).(*common.Config)
+
+	// Unpack the MSP config
+	rawMSPConfig := updatedConfig.ChannelGroup.Groups["Orderer"].Groups[orgID].Values["MSP"]
+	mspConfig := &msp.MSPConfig{}
+	err := proto.Unmarshal(rawMSPConfig.Value, mspConfig)
+	Expect(err).NotTo(HaveOccurred())
+
+	fabricConfig := &msp.FabricMSPConfig{}
+	err = proto.Unmarshal(mspConfig.Config, fabricConfig)
+	Expect(err).NotTo(HaveOccurred())
+
+	// Mutate it as we are asked
+	*fabricConfig = mutateMSP(*fabricConfig)
+
+	// Wrap it back into the config
+	mspConfig.Config = protoutil.MarshalOrPanic(fabricConfig)
+	rawMSPConfig.Value = protoutil.MarshalOrPanic(mspConfig)
+
+	UpdateOrdererConfig(network, orderer, channel, config, updatedConfig, peer, orderer)
+}