Assembler synchronizer

config/config.go

@@ -695,6 +695,26 @@ func (config *Configuration) ExtractConsenters() []nodeconfig.ConsenterInfo {
 	return consenters
 }
 
+func (config *Configuration) ExtractAssemblers() []nodeconfig.AssemblerInfo {
+	var assemblers []nodeconfig.AssemblerInfo
+	for _, party := range config.SharedConfig.PartiesConfig {
+		var tlsCACertsCollection []nodeconfig.RawBytes
+		for _, ca := range party.TLSCACerts {
+			tlsCACertsCollection = append(tlsCACertsCollection, ca)
+		}
+
+		assemblerInfo := nodeconfig.AssemblerInfo{
+			PartyID:    types.PartyID(party.PartyID),
+			Endpoint:   net.JoinHostPort(party.AssemblerConfig.Host, strconv.Itoa(int(party.AssemblerConfig.Port))),
+			TLSCACerts: tlsCACertsCollection,
+			TLSCert:    party.AssemblerConfig.TlsCert,
+		}
+		assemblers = append(assemblers, assemblerInfo)
+	}
+
+	return assemblers
+}
+
 func (config *Configuration) ExtractRouterInParty() nodeconfig.RouterInfo {
 	partyID := config.LocalConfig.NodeLocalConfig.PartyID
 	var party *ordererpb.PartyConfig
@@ -1050,3 +1070,30 @@ func ExtractConsenterAddresses(ordererConfig channelconfig.Orderer) (orderers.Pa
 
 	return party2Endpoint, nil
 }
+
+func ExtractAssemblerAddresses(ordererConfig channelconfig.Orderer) (orderers.Party2Endpoint, error) {
+	consensusMeta := ordererConfig.ConsensusMetadata()
+	sharedConfig := &ordererpb.SharedConfig{}
+	if err := proto.Unmarshal(consensusMeta, sharedConfig); err != nil {
+		return nil, errors.Wrap(err, "failed to unmarshal consensus metadata")
+	}
+
+	conf := &Configuration{
+		SharedConfig: sharedConfig,
+	}
+
+	aInfo := conf.ExtractAssemblers()
+
+	party2Endpoint := make(orderers.Party2Endpoint)
+
+	for _, assembler := range aInfo {
+		party2Endpoint[assembler.PartyID] = &orderers.Endpoint{
+			Address: assembler.Endpoint,
+		}
+		for _, cert := range assembler.TLSCACerts {
+			party2Endpoint[assembler.PartyID].RootCerts = append(party2Endpoint[assembler.PartyID].RootCerts, cert)
+		}
+	}
+
+	return party2Endpoint, nil
+}

node/assembler/synchronizer/block_verifier.go

@@ -0,0 +1,63 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package synchronizer
+
+import (
+	"github.com/hyperledger/fabric-lib-go/bccsp"
+	"github.com/hyperledger/fabric-lib-go/common/flogging"
+	"github.com/hyperledger/fabric-protos-go-apiv2/common"
+	"github.com/hyperledger/fabric-x-orderer/common/deliverclient"
+)
+
+//go:generate counterfeiter -o mocks/verifier_factory.go --fake-name VerifierFactory . VerifierFactory
+type VerifierFactory interface {
+	CreateBlockVerifier(
+		configBlock *common.Block,
+		lastBlock *common.Block,
+		cryptoProvider bccsp.BCCSP,
+		lg *flogging.FabricLogger,
+	) (deliverclient.CloneableUpdatableBlockVerifier, error)
+}
+
+// noopVerifierCreator creates a block verifier that does not actually verify blocks, which can be used in tests or when block verification is not needed.
+type noopVerifierCreator struct{}
+
+func (*noopVerifierCreator) CreateBlockVerifier(
+	configBlock *common.Block,
+	lastBlock *common.Block,
+	cryptoProvider bccsp.BCCSP,
+	lg *flogging.FabricLogger,
+) (deliverclient.CloneableUpdatableBlockVerifier, error) {
+	return &noopBlockVerifier{}, nil
+}
+
+// noopBlockVerifier is a block verifier that does not actually verify blocks, which can be used in tests or when block verification is not needed.
+type noopBlockVerifier struct{}
+
+// VerifyBlock checks block integrity and its relation to the chain, and verifies the signatures.
+func (*noopBlockVerifier) VerifyBlock(block *common.Block) error {
+	// TODO
+	return nil
+}
+
+func (*noopBlockVerifier) VerifyBlockAttestation(block *common.Block) error {
+	// TODO
+	return nil
+}
+
+func (*noopBlockVerifier) UpdateConfig(configBlock *common.Block) error {
+	// TODO
+	return nil
+}
+
+func (*noopBlockVerifier) UpdateBlockHeader(block *common.Block) {
+	// TODO
+}
+
+func (*noopBlockVerifier) Clone() deliverclient.CloneableUpdatableBlockVerifier {
+	return &noopBlockVerifier{}
+}

node/assembler/synchronizer/genesis_fetcher_factory.go

@@ -0,0 +1,148 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package synchronizer
+
+import (
+	"crypto/x509"
+	"encoding/pem"
+	"time"
+
+	"github.com/hyperledger/fabric-lib-go/bccsp"
+	"github.com/hyperledger/fabric-lib-go/common/flogging"
+	"github.com/hyperledger/fabric-protos-go-apiv2/common"
+	"github.com/hyperledger/fabric-x-orderer/common/types"
+	"github.com/hyperledger/fabric-x-orderer/config"
+	"github.com/hyperledger/fabric-x-orderer/node/comm"
+	"github.com/pkg/errors"
+)
+
+//go:generate counterfeiter -o mocks/genesis_fetcher.go . GenesisFetcher
+type GenesisFetcher interface {
+	GenesisByEndpoints() (map[string]*common.Block, error)
+	Close()
+}
+
+//go:generate counterfeiter -o mocks/genesis_fetcher_factory.go . GenesisFetcherFactory
+type GenesisFetcherFactory interface {
+	// CreateGenesisFetcher creates a new genesis fetcher.
+	CreateGenesisFetcher(
+		myPartyID types.PartyID,
+		support AssemblerSupport,
+		baseDialer *comm.PredicateDialer,
+		clusterConfig config.Cluster,
+		bccsp bccsp.BCCSP,
+		logger *flogging.FabricLogger,
+	) (GenesisFetcher, error)
+}
+
+type GenesisFetcherCreator struct{}
+
+func (*GenesisFetcherCreator) CreateGenesisFetcher(
+	myPartyID types.PartyID,
+	support AssemblerSupport,
+	baseDialer *comm.PredicateDialer,
+	clusterConfig config.Cluster,
+	bccsp bccsp.BCCSP,
+	logger *flogging.FabricLogger,
+) (GenesisFetcher, error) {
+	return newBlockPuller(myPartyID, support, baseDialer, clusterConfig, bccsp, logger)
+}
+
+// newBlockPuller creates a new block puller, which is used for target height detection.
+func newBlockPuller(
+	myPartyID types.PartyID,
+	support AssemblerSupport,
+	baseDialer *comm.PredicateDialer,
+	clusterConfig config.Cluster,
+	bccsp bccsp.BCCSP,
+	logger *flogging.FabricLogger,
+) (GenesisFetcher, error) {
+	// TODO replace this with the actual implementation
+	verifyBlockSequenceNoOp := func(blocks []*common.Block, _ string) error {
+		// TODO
+		return nil
+	}
+
+	stdDialer := &comm.StandardDialer{
+		Config: baseDialer.Config.Clone(),
+	}
+	stdDialer.Config.AsyncConnect = false
+	stdDialer.Config.SecOpts.VerifyCertificate = nil
+
+	// Extract endpoint and TLS cert from the config, excluding the self endpoint.
+	endpoints, err := extractEndpointCriteriaFromConfig(myPartyID, support)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to extract endpoint criteria from config")
+	}
+
+	der, _ := pem.Decode(stdDialer.Config.SecOpts.Certificate)
+	if der == nil {
+		return nil, errors.Errorf("client certificate isn't in PEM format: %v",
+			string(stdDialer.Config.SecOpts.Certificate))
+	}
+
+	myCert, err := x509.ParseCertificate(der.Bytes)
+	if err != nil {
+		logger.Warnf("Failed parsing my own TLS certificate: %v, therefore we may connect to our own endpoint when pulling blocks", err)
+	}
+
+	// TODO Fabric defaults. Extend the config to have these values, and use the config values instead of hardcoding them here.
+	// Cluster: Cluster{
+	// 	ReplicationMaxRetries:          12,
+	// 	RPCTimeout:                     time.Second * 7,
+	// 	DialTimeout:                    time.Second * 5,
+	// 	ReplicationBufferSize:          20971520,
+	// 	SendBufferSize:                 100,
+	// 	ReplicationRetryTimeout:        time.Second * 5,
+	// 	ReplicationPullTimeout:         time.Second * 5,
+	// 	CertExpirationWarningThreshold: time.Hour * 24 * 7,
+	// 	ReplicationPolicy:              "consensus", // BFT default; on etcdraft it is ignored
+	// },
+
+	bp := &comm.BlockPuller{
+		MyOwnTLSCert:        myCert,
+		VerifyBlockSequence: verifyBlockSequenceNoOp,
+		Logger:              logger,
+		RetryTimeout:        time.Second * 5,  // clusterConfig.ReplicationRetryTimeout,
+		MaxTotalBufferBytes: 20 * 1024 * 1024, // clusterConfig.ReplicationBufferSize,
+		FetchTimeout:        time.Second * 5,  // clusterConfig.ReplicationPullTimeout,
+		Endpoints:           endpoints,        // TODO the block puller is not party aware yet
+		Signer:              support,
+		TLSCert:             der.Bytes,
+		Channel:             support.ChannelID(),
+		Dialer:              stdDialer,
+	}
+
+	logger.Infof("Built new block puller (target height detector) with endpoints: %+v", endpoints)
+
+	return bp, nil
+}
+
+// extractEndpointCriteriaFromConfig extracts endpoint criteria from the channel configuration.
+// It retrieves all consenter addresses from the shared config and converts them into EndpointCriteria,
+// excluding the endpoint corresponding to myPartyID to avoid self-connection.
+// Returns a slice of EndpointCriteria containing the endpoint address and TLS root certificates for each consenter.
+func extractEndpointCriteriaFromConfig(myPartyID types.PartyID, support AssemblerSupport) ([]comm.EndpointCriteria, error) {
+	party2endpoint, err := config.ExtractAssemblerAddresses(support.SharedConfig())
+	if err != nil {
+		return nil, err
+	}
+
+	var endpoints []comm.EndpointCriteria
+	for party, ep := range party2endpoint {
+		if party == myPartyID {
+			continue
+		}
+		endpointCriteria := &comm.EndpointCriteria{
+			Endpoint:   ep.Address,
+			TLSRootCAs: ep.RootCerts,
+		}
+		endpoints = append(endpoints, *endpointCriteria)
+	}
+
+	return endpoints, nil
+}