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

When we expand the root TLS CA in the channel config, *after*
Raft membership has expanded with an OSN that is issed a certificate
by a new TLS CA, the TLS client handshake uses the old root CA pool
and as a result the added orderer cannot be reached by the existing ones,
because their dialers reject its certificate.

This change set builds a dynamic transport credentials that
re-computes the TLS config in every TLS client handshake.

Expanded an integration test to ensure this works.

Change-Id: I6578ba49f16e14b97eb4eef4feccdecbfe1b7015
Signed-off-by: yacovm <yacovm@il.ibm.com>

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"
 )
 
@@ -166,10 +165,13 @@ func (client *GRPCClient) SetServerRootCAs(serverRoots [][]byte) error {
 	return nil
 }
 
+// TLSOption changes the given TLS config
+type TLSOption func(tlsConfig *tls.Config)
+
 // NewConnection returns a grpc.ClientConn for the target address and
 // overrides the server name used to verify the hostname on the
 // certificate returned by a server when using TLS
-func (client *GRPCClient) NewConnection(address string, serverNameOverride string) (
+func (client *GRPCClient) NewConnection(address string, serverNameOverride string, tlsOptions ...TLSOption) (
 	*grpc.ClientConn, error) {
 
 	var dialOpts []grpc.DialOption
@@ -183,7 +185,7 @@ func (client *GRPCClient) NewConnection(address string, serverNameOverride strin
 		client.tlsConfig.ServerName = serverNameOverride
 		dialOpts = append(dialOpts,
 			grpc.WithTransportCredentials(
-				credentials.NewTLS(client.tlsConfig)))
+				&DynamicClientCredentials{TLSConfig: client.tlsConfig, TLSOptions: tlsOptions}))
 	} else {
 		dialOpts = append(dialOpts, grpc.WithInsecure())
 	}

core/comm/client_test.go

@@ -15,15 +15,20 @@ 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"
 	testpb "github.com/hyperledger/fabric/core/comm/testdata/grpc"
 	"github.com/pkg/errors"
 	"github.com/stretchr/testify/assert"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials"
 )
 
@@ -560,3 +565,86 @@ func loadCerts(t *testing.T) testCerts {
 
 	return certs
 }
+
+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

@@ -22,6 +22,8 @@ var (
 	OverrrideHostnameNotSupportedError = errors.New(
 		"core/comm: OverrideServerName is " +
 			"not supported")
+	ServerHandshakeNotImplementedError = errors.New("core/comm: server handshakes are not implemented with clientCreds")
+
 	MissingServerConfigError = errors.New(
 		"core/comm: `serverConfig` cannot be nil")
 	// alpnProtoStr are the specified application level protocols for gRPC.
@@ -89,3 +91,37 @@ func (sc *serverCreds) Clone() credentials.TransportCredentials {
 func (sc *serverCreds) OverrideServerName(string) error {
 	return OverrrideHostnameNotSupportedError
 }
+
+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, ServerHandshakeNotImplementedError
+}
+
+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/e2e/etcdraft_reconfig_test.go

@@ -8,6 +8,10 @@ package e2e
 
 import (
 	"bytes"
+	"crypto/ecdsa"
+	"crypto/rand"
+	"crypto/x509"
+	"encoding/pem"
 	"fmt"
 	"io/ioutil"
 	"os"
@@ -100,7 +104,7 @@ var _ = Describe("EndToEnd reconfiguration and onboarding", func() {
 	})
 
 	When("a single node cluster is expanded", func() {
-		It("is still possible to onboard the new cluster member", func() {
+		It("is still possible to onboard the new cluster member and then another one with a different TLS root CA", func() {
 			launch := func(o *nwo.Orderer) {
 				runner := network.OrdererRunner(o)
 				ordererRunners = append(ordererRunners, runner)
@@ -165,6 +169,108 @@ var _ = Describe("EndToEnd reconfiguration and onboarding", func() {
 			launch(orderer2)
 			By("Waiting for a leader to be re-elected")
 			findLeader(ordererRunners)
+
+			// In the next part of the test we're going to bring up a third node with a different TLS root CA
+			// and we're going to add the TLS root CA *after* we add it to the channel, to ensure
+			// that we can dynamically update TLS root CAs in Raft while membership stays the same.
+
+			By("Creating configuration for a third orderer with a different TLS root CA")
+			orderer3 := &nwo.Orderer{
+				Name:         "orderer3",
+				Organization: "OrdererOrg",
+			}
+			ports = nwo.Ports{}
+			for _, portName := range nwo.OrdererPortNames() {
+				ports[portName] = network.ReservePort()
+			}
+			network.PortsByOrdererID[orderer3.ID()] = ports
+			network.Orderers = append(network.Orderers, orderer3)
+			network.GenerateOrdererConfig(orderer3)
+
+			tmpDir, err := ioutil.TempDir("", "e2e-etcfraft_reconfig")
+			Expect(err).NotTo(HaveOccurred())
+			defer os.RemoveAll(tmpDir)
+
+			sess, err := network.Cryptogen(commands.Generate{
+				Config: network.CryptoConfigPath(),
+				Output: tmpDir,
+			})
+			Expect(err).NotTo(HaveOccurred())
+			Eventually(sess, network.EventuallyTimeout).Should(gexec.Exit(0))
+
+			name := network.Orderers[0].Name
+			domain := network.Organization(network.Orderers[0].Organization).Domain
+			nameDomain := fmt.Sprintf("%s.%s", name, domain)
+			ordererTLSPath := filepath.Join(tmpDir, "ordererOrganizations", domain, "orderers", nameDomain, "tls")
+
+			caCertPath := filepath.Join(tmpDir, "ordererOrganizations", domain, "tlsca", fmt.Sprintf("tlsca.%s-cert.pem", domain))
+			caCert, err := ioutil.ReadFile(caCertPath)
+			Expect(err).NotTo(HaveOccurred())
+
+			caKeyPath := filepath.Join(tmpDir, "ordererOrganizations", domain, "tlsca", privateKeyFileName(caCert))
+			caKey, err := ioutil.ReadFile(caKeyPath)
+			Expect(err).NotTo(HaveOccurred())
+
+			thirdOrdererCertificatePath := filepath.Join(ordererTLSPath, "server.crt")
+			thirdOrdererCertificate, err := ioutil.ReadFile(thirdOrdererCertificatePath)
+			Expect(err).NotTo(HaveOccurred())
+
+			By("Changing its subject name")
+			caCert, thirdOrdererCertificate = changeSubjectName(caCert, caKey, thirdOrdererCertificate, "tlsca2")
+
+			By("Updating it on the file system")
+			err = ioutil.WriteFile(caCertPath, caCert, 0644)
+			Expect(err).NotTo(HaveOccurred())
+			err = ioutil.WriteFile(thirdOrdererCertificatePath, thirdOrdererCertificate, 0644)
+			Expect(err).NotTo(HaveOccurred())
+
+			By("Overwriting the TLS directory of the new orderer")
+			for _, fileName := range []string{"server.crt", "server.key", "ca.crt"} {
+				dst := filepath.Join(network.OrdererLocalTLSDir(orderer3), fileName)
+
+				data, err := ioutil.ReadFile(filepath.Join(ordererTLSPath, fileName))
+				Expect(err).NotTo(HaveOccurred())
+
+				err = ioutil.WriteFile(dst, data, 0644)
+				Expect(err).NotTo(HaveOccurred())
+			}
+
+			By("Adding the third orderer to the channel")
+			nwo.AddConsenter(network, peer, orderer, "systemchannel", etcdraft.Consenter{
+				ServerTlsCert: thirdOrdererCertificate,
+				ClientTlsCert: thirdOrdererCertificate,
+				Host:          "127.0.0.1",
+				Port:          uint32(network.OrdererPort(orderer3, nwo.ClusterPort)),
+			})
+
+			By("Obtaining the last config block from the orderer once more to update the bootstrap file")
+			configBlock = nwo.GetConfigBlock(network, peer, orderer, "systemchannel")
+			err = ioutil.WriteFile(filepath.Join(testDir, "systemchannel_block.pb"), utils.MarshalOrPanic(configBlock), 0644)
+			Expect(err).NotTo(HaveOccurred())
+
+			By("Launching the third orderer")
+			launch(orderer3)
+
+			By("Expanding the TLS root CA certificates")
+			nwo.UpdateOrdererMSP(network, peer, orderer, "systemchannel", "OrdererOrg", func(config msp.FabricMSPConfig) msp.FabricMSPConfig {
+				config.TlsRootCerts = append(config.TlsRootCerts, caCert)
+				return config
+			})
+
+			By("Waiting for orderer3 to see the leader")
+			leader := findLeader([]*ginkgomon.Runner{ordererRunners[2]})
+			leaderIndex := leader - 1
+
+			fmt.Fprint(GinkgoWriter, "Killing the leader", leader)
+			ordererProcesses[leaderIndex].Signal(syscall.SIGTERM)
+			Eventually(ordererProcesses[leaderIndex].Wait(), network.EventuallyTimeout).Should(Receive())
+
+			By("Ensuring orderer3 detects leader loss")
+			leaderLoss := fmt.Sprintf("Raft leader changed: %d -> 0", leader)
+			Eventually(ordererRunners[2].Err(), network.EventuallyTimeout, time.Second).Should(gbytes.Say(leaderLoss))
+
+			By("Waiting for the leader to be re-elected")
+			findLeader([]*ginkgomon.Runner{ordererRunners[2]})
 		})
 	})
 
@@ -1384,3 +1490,44 @@ func revokeReaderAccess(network *nwo.Network, channel string, orderer *nwo.Order
 	updatedConfig.ChannelGroup.Groups["Orderer"].Policies["Readers"].Policy.Value = adminPolicy
 	nwo.UpdateOrdererConfig(network, orderer, channel, config, updatedConfig, peer, orderer)
 }
+
+func changeSubjectName(caCertPEM, caKeyPEM, leafPEM []byte, newSubjectName string) (newCA, newLeaf []byte) {
+	keyAsDER, _ := pem.Decode(caKeyPEM)
+	caKeyWithoutType, err := x509.ParsePKCS8PrivateKey(keyAsDER.Bytes)
+	Expect(err).NotTo(HaveOccurred())
+	caKey := caKeyWithoutType.(*ecdsa.PrivateKey)
+
+	caCertAsDER, _ := pem.Decode(caCertPEM)
+	caCert, err := x509.ParseCertificate(caCertAsDER.Bytes)
+	Expect(err).NotTo(HaveOccurred())
+
+	// Change its subject name
+	caCert.Subject.CommonName = newSubjectName
+	caCert.Issuer.CommonName = newSubjectName
+	caCert.RawTBSCertificate = nil
+	caCert.RawSubjectPublicKeyInfo = nil
+	caCert.Raw = nil
+	caCert.RawSubject = nil
+	caCert.RawIssuer = nil
+
+	// The CA signs its own certificate
+	caCertBytes, err := x509.CreateCertificate(rand.Reader, caCert, caCert, caCert.PublicKey, caKey)
+	Expect(err).NotTo(HaveOccurred())
+
+	// Now it's the turn of the leaf certificate
+	leafAsDER, _ := pem.Decode(leafPEM)
+	leafCert, err := x509.ParseCertificate(leafAsDER.Bytes)
+	Expect(err).NotTo(HaveOccurred())
+
+	leafCert.Raw = nil
+	leafCert.RawIssuer = nil
+	leafCert.RawTBSCertificate = nil
+
+	// The CA signs the leaf cert
+	leafCertBytes, err := x509.CreateCertificate(rand.Reader, leafCert, caCert, leafCert.PublicKey, caKey)
+	Expect(err).NotTo(HaveOccurred())
+
+	newCA = pem.EncodeToMemory(&pem.Block{Bytes: caCertBytes, Type: "CERTIFICATE"})
+	newLeaf = pem.EncodeToMemory(&pem.Block{Bytes: leafCertBytes, Type: "CERTIFICATE"})
+	return
+}