add configStreamer to router

common/utils/pem_utils.go

@@ -0,0 +1,72 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package utils
+
+import (
+	"crypto/ecdsa"
+	"crypto/x509"
+	"encoding/pem"
+	"fmt"
+	"os"
+)
+
+func ReadPem(path string) ([]byte, error) {
+	if path == "" {
+		return nil, fmt.Errorf("failed reading pem file, path is empty")
+	}
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return nil, fmt.Errorf("failed reading a pem file from %s, err: %v", path, err)
+	}
+
+	return data, nil
+}
+
+func blockToPublicKey(block *pem.Block) []byte {
+	cert, err := x509.ParseCertificate(block.Bytes)
+	if err != nil {
+		panic(fmt.Sprintf("Failed parsing consenter signing certificate: %v", err))
+	}
+
+	pubKey, ok := cert.PublicKey.(*ecdsa.PublicKey)
+	if !ok {
+		panic(fmt.Sprintf("Failed parsing consenter public key: %v", err))
+	}
+
+	publicKeyBytes, err := x509.MarshalPKIXPublicKey(pubKey)
+	if err != nil {
+		panic(fmt.Sprintf("Failed marshaling consenter public key: %v", err))
+	}
+
+	pemPublicKey := pem.EncodeToMemory(&pem.Block{
+		Type:  "PUBLIC KEY",
+		Bytes: publicKeyBytes,
+	})
+
+	return pemPublicKey
+}
+
+func GetPublicKeyFromCertificate(nodeCert []byte) []byte {
+	// Fetch public key from signing certificate
+	// NOTE: ARMA's new configuration now uses certificates, which inherently contain the public key, instead of a separate public key field.
+	// To ensure backward compatibility until the full new config integration, the public key it enabled.
+	block, _ := pem.Decode(nodeCert)
+	if block == nil || block.Bytes == nil {
+		panic("Failed decoding consenter signing certificate")
+	}
+
+	var pemPublicKey []byte
+	if block.Type == "CERTIFICATE" {
+		pemPublicKey = blockToPublicKey(block)
+	}
+
+	if block.Type == "PUBLIC KEY" {
+		pemPublicKey = nodeCert
+	}
+
+	return pemPublicKey
+}

config/config.go

@@ -7,9 +7,6 @@ SPDX-License-Identifier: Apache-2.0
 package config
 
 import (
-	"crypto/ecdsa"
-	"crypto/x509"
-	"encoding/pem"
 	"fmt"
 	"os"
 	"path/filepath"
@@ -151,6 +148,7 @@ func (config *Configuration) ExtractRouterConfig() *nodeconfig.RouterNodeConfig
 		TLSPrivateKeyFile:                   config.LocalConfig.TLSConfig.PrivateKey,
 		ListenAddress:                       config.LocalConfig.NodeLocalConfig.GeneralConfig.ListenAddress + ":" + strconv.Itoa(int(config.LocalConfig.NodeLocalConfig.GeneralConfig.ListenPort)),
 		Shards:                              config.ExtractShards(),
+		Consenter:                           config.ExtractConsenterInParty(),
 		NumOfConnectionsForBatcher:          config.LocalConfig.NodeLocalConfig.RouterParams.NumberOfConnectionsPerBatcher,
 		NumOfgRPCStreamsPerConnection:       config.LocalConfig.NodeLocalConfig.RouterParams.NumberOfStreamsPerConnection,
 		UseTLS:                              config.LocalConfig.TLSConfig.Enabled,
@@ -216,6 +214,7 @@ func (config *Configuration) ExtractConsenterConfig() *nodeconfig.ConsenterNodeC
 	consenterConfig := &nodeconfig.ConsenterNodeConfig{
 		Shards:             config.ExtractShards(),
 		Consenters:         config.ExtractConsenters(),
+		Router:             config.ExtractRouterInParty(),
 		Directory:          config.LocalConfig.NodeLocalConfig.FileStore.Path,
 		ListenAddress:      config.LocalConfig.NodeLocalConfig.GeneralConfig.ListenAddress + ":" + strconv.Itoa(int(config.LocalConfig.NodeLocalConfig.GeneralConfig.ListenPort)),
 		PartyId:            config.LocalConfig.NodeLocalConfig.PartyID,
@@ -270,22 +269,7 @@ func (config *Configuration) ExtractShards() []nodeconfig.ShardInfo {
 		for _, batcher := range party.BatchersConfig {
 			shardId := types.ShardID(batcher.ShardID)
 
-			// Fetch public key from signing certificate
-			// NOTE: ARMA's new configuration uses certificates, which inherently contain the public key, instead of a separate public key field.
-			// To ensure backward compatibility until the full new config integration, the public key it enabled.
-			block, _ := pem.Decode(batcher.SignCert)
-			if block == nil || block.Bytes == nil {
-				panic("Failed decoding batcher signing certificate")
-			}
-
-			var pemPublicKey []byte
-			if block.Type == "CERTIFICATE" {
-				pemPublicKey = blockToPublicKey(block)
-			}
-
-			if block.Type == "PUBLIC KEY" {
-				pemPublicKey = batcher.SignCert
-			}
+			pemPublicKey := utils.GetPublicKeyFromCertificate(batcher.SignCert)
 
 			batcher := nodeconfig.BatcherInfo{
 				PartyID:    types.PartyID(party.PartyID),
@@ -323,22 +307,7 @@ func (config *Configuration) ExtractConsenters() []nodeconfig.ConsenterInfo {
 			tlsCACertsCollection = append(tlsCACertsCollection, ca)
 		}
 
-		// Fetch public key from signing certificate
-		// NOTE: ARMA's new configuration now uses certificates, which inherently contain the public key, instead of a separate public key field.
-		// To ensure backward compatibility until the full new config integration, the public key it enabled.
-		block, _ := pem.Decode(party.ConsenterConfig.SignCert)
-		if block == nil || block.Bytes == nil {
-			panic("Failed decoding consenter signing certificate")
-		}
-
-		var pemPublicKey []byte
-		if block.Type == "CERTIFICATE" {
-			pemPublicKey = blockToPublicKey(block)
-		}
-
-		if block.Type == "PUBLIC KEY" {
-			pemPublicKey = party.ConsenterConfig.SignCert
-		}
+		pemPublicKey := utils.GetPublicKeyFromCertificate(party.ConsenterConfig.SignCert)
 
 		consenterInfo := nodeconfig.ConsenterInfo{
 			PartyID:    types.PartyID(party.PartyID),
@@ -352,26 +321,28 @@ func (config *Configuration) ExtractConsenters() []nodeconfig.ConsenterInfo {
 	return consenters
 }
 
-func blockToPublicKey(block *pem.Block) []byte {
-	cert, err := x509.ParseCertificate(block.Bytes)
-	if err != nil {
-		panic(fmt.Sprintf("Failed parsing consenter signing certificate: %v", err))
-	}
+func (config *Configuration) ExtractRouterInParty() nodeconfig.RouterInfo {
+	partyID := config.LocalConfig.NodeLocalConfig.PartyID
+	party := config.SharedConfig.PartiesConfig[partyID-1]
+	routerConfig := party.RouterConfig
 
-	pubKey, ok := cert.PublicKey.(*ecdsa.PublicKey)
-	if !ok {
-		panic(fmt.Sprintf("Failed parsing consenter public key: %v", err))
+	var tlsCACertsCollection []nodeconfig.RawBytes
+	for _, ca := range party.TLSCACerts {
+		tlsCACertsCollection = append(tlsCACertsCollection, ca)
 	}
 
-	publicKeyBytes, err := x509.MarshalPKIXPublicKey(pubKey)
-	if err != nil {
-		panic(fmt.Sprintf("Failed marshaling consenter public key: %v", err))
+	routerInfo := nodeconfig.RouterInfo{
+		PartyID:    partyID,
+		Endpoint:   routerConfig.Host + ":" + strconv.Itoa(int(routerConfig.Port)),
+		TLSCACerts: tlsCACertsCollection,
+		TLSCert:    routerConfig.TlsCert,
 	}
 
-	pemPublicKey := pem.EncodeToMemory(&pem.Block{
-		Type:  "PUBLIC KEY",
-		Bytes: publicKeyBytes,
-	})
+	return routerInfo
+}
 
-	return pemPublicKey
+func (config *Configuration) ExtractConsenterInParty() nodeconfig.ConsenterInfo {
+	partyID := config.LocalConfig.NodeLocalConfig.PartyID
+	consenterInfos := config.ExtractConsenters()
+	return consenterInfos[partyID-1]
 }

node/batcher/stub_consenter_test.go

@@ -7,6 +7,7 @@ SPDX-License-Identifier: Apache-2.0
 package batcher_test
 
 import (
+	"context"
 	"fmt"
 	"io"
 	"sync"
@@ -81,6 +82,10 @@ func (sc *stubConsenter) NotifyEvent(stream protos.Consensus_NotifyEventServer)
 	}
 }
 
+func (sc *stubConsenter) SubmitConfig(ctx context.Context, request *protos.Request) (*protos.SubmitResponse, error) {
+	return nil, fmt.Errorf("not implemented")
+}
+
 func (sc *stubConsenter) Stop() {
 	// Stop() of stub consenter does nothing
 	// use NetStop() to stop the stub consenter network

node/config/config.go

@@ -65,6 +65,13 @@ type ConsenterInfo struct {
 	TLSCACerts []RawBytes
 }
 
+type RouterInfo struct {
+	PartyID    types.PartyID
+	Endpoint   string
+	TLSCACerts []RawBytes
+	TLSCert    RawBytes
+}
+
 type RouterNodeConfig struct {
 	// Private config
 	PartyID            types.PartyID
@@ -73,6 +80,7 @@ type RouterNodeConfig struct {
 	ListenAddress      string
 	// Shared config
 	Shards                              []ShardInfo
+	Consenter                           ConsenterInfo
 	NumOfConnectionsForBatcher          int
 	NumOfgRPCStreamsPerConnection       int
 	UseTLS                              bool
@@ -130,6 +138,7 @@ type ConsenterNodeConfig struct {
 	// Shared config
 	Shards        []ShardInfo
 	Consenters    []ConsenterInfo
+	Router        RouterInfo
 	Directory     string
 	ListenAddress string
 	// Private config

node/config/config_test.go

@@ -24,13 +24,15 @@ func TestRouterNodeConfigToYaml(t *testing.T) {
 		{1, "127.0.0.1:7050", []RawBytes{{1, 2, 3}, {4, 5, 6}}, RawBytes("BatcherPubKey-1"), RawBytes("TLS CERT")},
 		{2, "127.0.0.1:7051", []RawBytes{{1, 2, 3}, {4, 5, 6}}, RawBytes("BatcherPubKey-2"), RawBytes("TLS CERT")},
 	}
+	consenter := ConsenterInfo{1, "127.0.0.1:7050", RawBytes("ConsenterPubKey-1"), []RawBytes{{1, 2, 3}, {4, 5, 6}}}
 
 	shards := []ShardInfo{{ShardId: 1, Batchers: batchers}}
 	rnc := &RouterNodeConfig{
 		TLSCertificateFile:            []byte("tls cert"),
 		TLSPrivateKeyFile:             []byte("tls key"),
 		PartyID:                       1,
 		Shards:                        shards,
+		Consenter:                     consenter,
 		NumOfConnectionsForBatcher:    1,
 		NumOfgRPCStreamsPerConnection: 2,
 	}
@@ -87,10 +89,12 @@ func TestConsenterNodeConfigToYaml(t *testing.T) {
 	}
 	shards := []ShardInfo{{ShardId: 1, Batchers: batchers}}
 	consenters := []ConsenterInfo{{1, "127.0.0.1:7050", RawBytes("ConsenterPubKey-1"), []RawBytes{{1, 2, 3}, {4, 5, 6}}}}
+	router := RouterInfo{1, "127.0.0.1:7050", []RawBytes{{1, 2, 3}, {4, 5, 6}}, RawBytes("ConsenterPubKey-1")}
 
 	cnc := &ConsenterNodeConfig{
 		Shards:             shards,
 		Consenters:         consenters,
+		Router:             router,
 		PartyId:            1,
 		TLSPrivateKeyFile:  RawBytes("TlsPrivateKey"),
 		TLSCertificateFile: RawBytes("TlsCertKey"),

node/consensus/consensus.go

@@ -8,8 +8,10 @@ package consensus
 
 import (
 	"bytes"
+	"context"
 	"encoding/asn1"
 	"encoding/base64"
+	"encoding/pem"
 	"fmt"
 	"io"
 	"math"
@@ -22,6 +24,7 @@ import (
 	"github.com/hyperledger/fabric-protos-go-apiv2/common"
 	"github.com/hyperledger/fabric-protos-go-apiv2/orderer"
 	arma_types "github.com/hyperledger/fabric-x-orderer/common/types"
+	"github.com/hyperledger/fabric-x-orderer/node"
 	"github.com/hyperledger/fabric-x-orderer/node/comm"
 	"github.com/hyperledger/fabric-x-orderer/node/config"
 	"github.com/hyperledger/fabric-x-orderer/node/consensus/badb"
@@ -523,3 +526,35 @@ func (c *Consensus) pickEndpoint() string {
 	c.Logger.Debugf("Returning random node (ID=%d) endpoint : %s", c.Config.Consenters[r].PartyID, c.Config.Consenters[r].Endpoint)
 	return c.Config.Consenters[r].Endpoint
 }
+
+func (c *Consensus) SubmitConfig(ctx context.Context, request *protos.Request) (*protos.SubmitResponse, error) {
+	err := c.validateRouterFromContext(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	c.Logger.Debugf("Received config request from router %s", c.Config.Router.Endpoint)
+
+	return &protos.SubmitResponse{Error: "SubmitConfig is not implemented in consenter", TraceId: request.TraceId}, nil
+}
+
+func (c *Consensus) validateRouterFromContext(ctx context.Context) error {
+	// extract the client certificate from the context
+	cert := node.ExtractCertificateFromContext(ctx)
+	if cert == nil {
+		return errors.New("error: access denied; could not extract certificate from context")
+	}
+
+	// extract the router certificate from the ConsenterNodeConfig
+	rawRouterCert := c.Config.Router.TLSCert
+	pemBlock, _ := pem.Decode(rawRouterCert)
+	if pemBlock == nil || pemBlock.Bytes == nil {
+		return fmt.Errorf("error decoding router TLS certificate")
+	}
+
+	// compare the two certificates
+	if !bytes.Equal(pemBlock.Bytes, cert.Raw) {
+		return fmt.Errorf("error: access denied; client certificatte is different than the router's certificate")
+	}
+	return nil
+}