[FAB-12502] Deliver client support for etcdraft

This change set prepares the block puller to be attached
to the etcdraft consenter and chain.

It adds the needed configuration in the orderer.yaml,
and also adds code that creates a block puller.

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

Author: yacovm

orderer/common/cluster/util.go

@@ -10,6 +10,7 @@ import (
 	"bytes"
 	"encoding/hex"
 	"encoding/pem"
+	"reflect"
 	"sync/atomic"
 
 	"github.com/hyperledger/fabric/common/channelconfig"
@@ -106,6 +107,31 @@ func NewTLSPinningDialer(config comm.ClientConfig) *PredicateDialer {
 	return d
 }
 
+// ClientConfig returns the comm.ClientConfig, or an error
+// if they cannot be extracted.
+func (dialer *PredicateDialer) ClientConfig() (comm.ClientConfig, error) {
+	val := dialer.Config.Load()
+	if val == nil {
+		return comm.ClientConfig{}, errors.New("client config not initialized")
+	}
+	if cc, isClientConfig := val.(comm.ClientConfig); !isClientConfig {
+		err := errors.Errorf("value stored is %v, not comm.ClientConfig",
+			reflect.TypeOf(val))
+		return comm.ClientConfig{}, err
+	} else {
+		if cc.SecOpts == nil {
+			return comm.ClientConfig{}, errors.New("SecOpts is nil")
+		}
+		// Copy by value the secure options
+		secOpts := *cc.SecOpts
+		return comm.ClientConfig{
+			Timeout: cc.Timeout,
+			SecOpts: &secOpts,
+			KaOpts:  cc.KaOpts,
+		}, nil
+	}
+}
+
 // SetConfig sets the configuration of the PredicateDialer
 func (dialer *PredicateDialer) SetConfig(config comm.ClientConfig) {
 	configCopy := comm.ClientConfig{
@@ -147,13 +173,13 @@ func DERtoPEM(der []byte) string {
 	}))
 }
 
-// StandardDialerDialer wraps a PredicateDialer
+// StandardDialer wraps a PredicateDialer
 // to a standard cluster.Dialer that passes in a nil verify function
-type StandardDialerDialer struct {
+type StandardDialer struct {
 	Dialer *PredicateDialer
 }
 
-func (bdp *StandardDialerDialer) Dial(address string) (*grpc.ClientConn, error) {
+func (bdp *StandardDialer) Dial(address string) (*grpc.ClientConn, error) {
 	return bdp.Dialer.Dial(address, nil)
 }
 
@@ -264,17 +290,24 @@ func VerifyBlockSignature(block *common.Block, verifier BlockVerifier) error {
 	return verifier.VerifyBlockSignature(signatureSet)
 }
 
-// TLSCACertsFromConfigBlock retrieves TLS CA certificates
+// EndpointConfig defines a configuration
+// of endpoints of ordering service nodes
+type EndpointConfig struct {
+	TLSRootCAs [][]byte
+	Endpoints  []string
+}
+
+// EndpointconfigFromConfigBlock retrieves TLS CA certificates and endpoints
 // from a config block.
-func TLSCACertsFromConfigBlock(block *common.Block) ([][]byte, error) {
+func EndpointconfigFromConfigBlock(block *common.Block) (*EndpointConfig, error) {
 	if block == nil {
 		return nil, errors.New("nil block")
 	}
 	envelopeConfig, err := utils.ExtractEnvelope(block, 0)
 	if err != nil {
 		return nil, err
 	}
-	var res [][]byte
+	var tlsCACerts [][]byte
 	bundle, err := channelconfig.NewBundleFromEnvelope(envelopeConfig)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed extracting bundle from envelope")
@@ -292,7 +325,10 @@ func TLSCACertsFromConfigBlock(block *common.Block) ([][]byte, error) {
 		if msp == nil {
 			return nil, errors.Errorf("no MSP found for MSP with ID of %s", org.MSPID())
 		}
-		res = append(res, msp.GetTLSRootCerts()...)
+		tlsCACerts = append(tlsCACerts, msp.GetTLSRootCerts()...)
 	}
-	return res, nil
+	return &EndpointConfig{
+		Endpoints:  bundle.ChannelConfig().OrdererAddresses(),
+		TLSRootCAs: tlsCACerts,
+	}, nil
 }

orderer/common/cluster/util_test.go

@@ -105,7 +105,7 @@ func TestStandardDialerDialer(t *testing.T) {
 	t.Parallel()
 	emptyCertificate := []byte("-----BEGIN CERTIFICATE-----\n-----END CERTIFICATE-----")
 	dialer := cluster.NewTLSPinningDialer(comm.ClientConfig{SecOpts: &comm.SecureOptions{UseTLS: true, ServerRootCAs: [][]byte{emptyCertificate}}})
-	standardDialer := &cluster.StandardDialerDialer{Dialer: dialer}
+	standardDialer := &cluster.StandardDialer{Dialer: dialer}
 	_, err := standardDialer.Dial("127.0.0.1:8080")
 	assert.EqualError(t, err, "error adding root certificate: asn1: syntax error: sequence truncated")
 }
@@ -332,48 +332,49 @@ func createBlockChain(start, end uint64) []*common.Block {
 	return blockchain
 }
 
-func TestTLSCACertsFromConfigBlockGreenPath(t *testing.T) {
+func TestEndpointconfigFromConfigBlockGreenPath(t *testing.T) {
 	blockBytes, err := ioutil.ReadFile("testdata/mychannel.block")
 	assert.NoError(t, err)
 
 	block := &common.Block{}
 	assert.NoError(t, proto.Unmarshal(blockBytes, block))
 
-	certs, err := cluster.TLSCACertsFromConfigBlock(block)
+	endpointConfig, err := cluster.EndpointconfigFromConfigBlock(block)
 	assert.NoError(t, err)
-	assert.Len(t, certs, 1)
+	assert.Len(t, endpointConfig.TLSRootCAs, 1)
+	assert.Equal(t, []string{"orderer.example.com:7050"}, endpointConfig.Endpoints)
 
-	bl, _ := pem.Decode(certs[0])
+	bl, _ := pem.Decode(endpointConfig.TLSRootCAs[0])
 	cert, err := x509.ParseCertificate(bl.Bytes)
 	assert.NoError(t, err)
 
 	assert.True(t, cert.IsCA)
 	assert.Equal(t, "tlsca.example.com", cert.Subject.CommonName)
 }
 
-func TestTLSCACertsFromConfigBlockFailures(t *testing.T) {
+func TestEndpointconfigFromConfigBlockFailures(t *testing.T) {
 	t.Run("nil block", func(t *testing.T) {
-		certs, err := cluster.TLSCACertsFromConfigBlock(nil)
+		certs, err := cluster.EndpointconfigFromConfigBlock(nil)
 		assert.Nil(t, certs)
 		assert.EqualError(t, err, "nil block")
 	})
 
 	t.Run("nil block data", func(t *testing.T) {
-		certs, err := cluster.TLSCACertsFromConfigBlock(&common.Block{})
+		certs, err := cluster.EndpointconfigFromConfigBlock(&common.Block{})
 		assert.Nil(t, certs)
 		assert.EqualError(t, err, "block data is nil")
 	})
 
 	t.Run("no envelope", func(t *testing.T) {
-		certs, err := cluster.TLSCACertsFromConfigBlock(&common.Block{
+		certs, err := cluster.EndpointconfigFromConfigBlock(&common.Block{
 			Data: &common.BlockData{},
 		})
 		assert.Nil(t, certs)
 		assert.EqualError(t, err, "envelope index out of bounds")
 	})
 
 	t.Run("bad envelope", func(t *testing.T) {
-		certs, err := cluster.TLSCACertsFromConfigBlock(&common.Block{
+		certs, err := cluster.EndpointconfigFromConfigBlock(&common.Block{
 			Data: &common.BlockData{
 				Data: [][]byte{{}},
 			},
@@ -382,3 +383,39 @@ func TestTLSCACertsFromConfigBlockFailures(t *testing.T) {
 		assert.EqualError(t, err, "failed extracting bundle from envelope: envelope header cannot be nil")
 	})
 }
+
+func TestClientConfig(t *testing.T) {
+	t.Run("Uninitialized dialer", func(t *testing.T) {
+		dialer := &cluster.PredicateDialer{}
+		_, err := dialer.ClientConfig()
+		assert.EqualError(t, err, "client config not initialized")
+	})
+
+	t.Run("Wrong type stored", func(t *testing.T) {
+		dialer := &cluster.PredicateDialer{}
+		dialer.Config.Store("foo")
+		_, err := dialer.ClientConfig()
+		assert.EqualError(t, err, "value stored is string, not comm.ClientConfig")
+	})
+
+	t.Run("Nil secure options", func(t *testing.T) {
+		dialer := &cluster.PredicateDialer{}
+		dialer.Config.Store(comm.ClientConfig{
+			SecOpts: nil,
+		})
+		_, err := dialer.ClientConfig()
+		assert.EqualError(t, err, "SecOpts is nil")
+	})
+
+	t.Run("Valid config", func(t *testing.T) {
+		dialer := &cluster.PredicateDialer{}
+		dialer.Config.Store(comm.ClientConfig{
+			SecOpts: &comm.SecureOptions{
+				Key: []byte{1, 2, 3},
+			},
+		})
+		cc, err := dialer.ClientConfig()
+		assert.NoError(t, err)
+		assert.Equal(t, []byte{1, 2, 3}, cc.SecOpts.Key)
+	})
+}

orderer/common/localconfig/config.go

@@ -57,11 +57,14 @@ type General struct {
 }
 
 type Cluster struct {
-	RootCAs           []string
-	ClientCertificate string
-	ClientPrivateKey  string
-	DialTimeout       time.Duration
-	RPCTimeout        time.Duration
+	RootCAs                 []string
+	ClientCertificate       string
+	ClientPrivateKey        string
+	DialTimeout             time.Duration
+	RPCTimeout              time.Duration
+	ReplicationBufferSize   int
+	ReplicationPullTimeout  time.Duration
+	ReplicationRetryTimeout time.Duration
 }
 
 // Keepalive contains configuration for gRPC servers.

orderer/common/multichannel/chainsupport.go

@@ -9,6 +9,7 @@ package multichannel
 import (
 	"github.com/hyperledger/fabric/common/crypto"
 	"github.com/hyperledger/fabric/common/ledger/blockledger"
+	"github.com/hyperledger/fabric/common/policies"
 	"github.com/hyperledger/fabric/orderer/common/blockcutter"
 	"github.com/hyperledger/fabric/orderer/common/msgprocessor"
 	"github.com/hyperledger/fabric/orderer/consensus"
@@ -138,3 +139,16 @@ func (cs *ChainSupport) ConfigProto() *cb.Config {
 func (cs *ChainSupport) Sequence() uint64 {
 	return cs.ConfigtxValidator().Sequence()
 }
+
+// VerifyBlockSignature verifies a signature of a block
+func (cs *ChainSupport) VerifyBlockSignature(sd []*cb.SignedData) error {
+	policy, exists := cs.PolicyManager().GetPolicy(policies.BlockValidation)
+	if !exists {
+		return errors.Errorf("policy %s wasn't found", policies.BlockValidation)
+	}
+	err := policy.Evaluate(sd)
+	if err != nil {
+		return errors.Wrap(err, "block verification failed")
+	}
+	return nil
+}

orderer/common/multichannel/chainsupprt_test.go

@@ -9,10 +9,15 @@ package multichannel
 import (
 	"testing"
 
+	"github.com/hyperledger/fabric/common/channelconfig"
 	"github.com/hyperledger/fabric/common/deliver/mock"
 	"github.com/hyperledger/fabric/common/ledger/blockledger/mocks"
+	"github.com/hyperledger/fabric/common/mocks/config"
+	mockpolicies "github.com/hyperledger/fabric/common/mocks/policies"
+	"github.com/hyperledger/fabric/common/policies"
 	"github.com/hyperledger/fabric/protos/common"
 	"github.com/hyperledger/fabric/protos/orderer"
+	"github.com/pkg/errors"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -31,3 +36,48 @@ func TestChainSupportBlock(t *testing.T) {
 	assert.Nil(t, cs.Block(100))
 	assert.Equal(t, uint64(99), cs.Block(99).Header.Number)
 }
+
+type mutableResourcesMock struct {
+	config.Resources
+}
+
+func (*mutableResourcesMock) Update(*channelconfig.Bundle) {
+	panic("implement me")
+}
+
+func TestVerifyBlockSignature(t *testing.T) {
+	policyMgr := &mockpolicies.Manager{
+		PolicyMap: make(map[string]policies.Policy),
+	}
+	ms := &mutableResourcesMock{
+		Resources: config.Resources{
+			PolicyManagerVal: policyMgr,
+		},
+	}
+	cs := &ChainSupport{
+		ledgerResources: &ledgerResources{
+			configResources: &configResources{
+				mutableResources: ms,
+			},
+		},
+	}
+
+	// Scenario I: Policy manager isn't initialized
+	// and thus policy cannot be found
+	err := cs.VerifyBlockSignature([]*common.SignedData{})
+	assert.EqualError(t, err, "policy /Channel/Orderer/BlockValidation wasn't found")
+
+	// Scenario II: Policy manager finds policy, but it evaluates
+	// to error.
+	policyMgr.PolicyMap["/Channel/Orderer/BlockValidation"] = &mockpolicies.Policy{
+		Err: errors.New("invalid signature"),
+	}
+	err = cs.VerifyBlockSignature([]*common.SignedData{})
+	assert.EqualError(t, err, "block verification failed: invalid signature")
+
+	// Scenario III: Policy manager finds policy, and it evaluates to success
+	policyMgr.PolicyMap["/Channel/Orderer/BlockValidation"] = &mockpolicies.Policy{
+		Err: nil,
+	}
+	assert.NoError(t, cs.VerifyBlockSignature([]*common.SignedData{}))
+}

orderer/consensus/consensus.go

@@ -74,6 +74,9 @@ type ConsenterSupport interface {
 	crypto.LocalSigner
 	msgprocessor.Processor
 
+	// VerifyBlockSignature verifies a signature of a block
+	VerifyBlockSignature([]*cb.SignedData) error
+
 	// BlockCutter returns the block cutting helper for this channel.
 	BlockCutter() blockcutter.Receiver