Verify signed transactions

Signed-off-by: Genady Gurevich <[email protected]>

Author: gengur

common/tools/armageddon/armageddon.go

@@ -352,7 +352,7 @@ func generateConfigAndCrypto(genConfigFile **os.File, outputDir *string, sampleC
 		userTLSPrivateKeyPath := filepath.Join(*outputDir, "crypto", "ordererOrganizations", fmt.Sprintf("org%d", i+1), "users", "user", "tls", "user-key.pem")
 		userTLSCertPath := filepath.Join(*outputDir, "crypto", "ordererOrganizations", fmt.Sprintf("org%d", i+1), "users", "user", "tls", "user-tls-cert.pem")
 
-		userConfig, err := NewUserConfig(userTLSPrivateKeyPath, userTLSCertPath, tlsCACertsBytesPartiesCollection, networkConfig)
+		userConfig, err := NewUserConfig(*outputDir, userTLSPrivateKeyPath, userTLSCertPath, tlsCACertsBytesPartiesCollection, networkConfig)
 		if err != nil {
 			fmt.Fprintf(os.Stderr, "Error creating user config: %s", err)
 			os.Exit(-1)
@@ -596,9 +596,9 @@ func sendTxToRouters(userConfig *UserConfig, numOfTxs int, rate int, txSize int,
 	var streams []ab.AtomicBroadcast_BroadcastClient
 
 	// create gRPC clients and streams to the routers
-	for i := 0; i < len(userConfig.RouterEndpoints); i++ {
+	for i := 0; i < len(userConfig.RouterUserConfigs); i++ {
 		// create a gRPC connection to the router
-		gRPCRouterClientConn, stream, err := createConnAndStream(userConfig, userConfig.RouterEndpoints[i])
+		gRPCRouterClientConn, stream, err := createConnAndStream(userConfig, userConfig.RouterUserConfigs[i].Endpoint)
 		if err != nil {
 			fmt.Fprintf(os.Stderr, "failed to create a gRPC client connection and stream to router %d, err: %v", i+1, err)
 			os.Exit(3)

common/tools/armageddon/broadcast_client.go

@@ -128,14 +128,14 @@ type BroadcastTxClient struct {
 func NewBroadcastTxClient(userConfigFile *UserConfig) *BroadcastTxClient {
 	return &BroadcastTxClient{
 		userConfig:       userConfigFile,
-		streamsToRouters: make([]*StreamInfo, len(userConfigFile.RouterEndpoints)),
+		streamsToRouters: make([]*StreamInfo, len(userConfigFile.RouterUserConfigs)),
 		stopChan:         make(chan struct{}),
 	}
 }
 
 func (c *BroadcastTxClient) InitStreams() error {
-	for i, routerEndpoint := range c.userConfig.RouterEndpoints {
-		conn, stream, err := createConnAndStream(c.userConfig, routerEndpoint)
+	for i, routerUserConfig := range c.userConfig.RouterUserConfigs {
+		conn, stream, err := createConnAndStream(c.userConfig, routerUserConfig.Endpoint)
 		if err != nil {
 			return err
 		}
@@ -145,7 +145,7 @@ func (c *BroadcastTxClient) InitStreams() error {
 			stream:        stream,
 			stopChan:      make(chan struct{}),
 			maxRetryDelay: 8 * time.Second,
-			endpoint:      routerEndpoint,
+			endpoint:      routerUserConfig.Endpoint,
 			lock:          sync.Mutex{},
 			logger:        flogging.MustGetLogger(fmt.Sprintf("BroadcastClientToRouter%d", i+1)),
 		}

common/tools/armageddon/user.go

@@ -8,29 +8,41 @@ package armageddon
 
 import (
 	"fmt"
+	"path/filepath"
 
+	"github.com/hyperledger/fabric-x-orderer/common/types"
 	"github.com/hyperledger/fabric-x-orderer/common/utils"
 	genconfig "github.com/hyperledger/fabric-x-orderer/config/generate"
 )
 
+type RouterUserConfig struct {
+	Endpoint   string        `yaml:"Endpoint,omitempty"`
+	PartyID    types.PartyID `yaml:"PartyID,omitempty"`
+	UserMSPDir string        `yaml:"UserMSPDir,omitempty"`
+}
+
 // UserConfig holds the user information needed for connection to routers and assemblers
 // Note: a user will be created for each party. One of the users will be chosen as a grpc client that sends tx to all router and receives blocks from the assemblers.
 type UserConfig struct {
-	TLSPrivateKey      []byte   `yaml:"TLSPrivateKey,omitempty"`
-	TLSCertificate     []byte   `yaml:"TLSCertificate,omitempty"`
-	RouterEndpoints    []string `yaml:"RouterEndpoints,omitempty"`
-	AssemblerEndpoints []string `yaml:"AssemblerEndpoints,omitempty"`
-	TLSCACerts         [][]byte `yaml:"TLSCACerts,omitempty"`
-	UseTLSRouter       string   `yaml:"UseTLSRouter,omitempty"`
-	UseTLSAssembler    string   `yaml:"UseTLSAssembler,omitempty"`
+	TLSPrivateKey      []byte             `yaml:"TLSPrivateKey,omitempty"`
+	TLSCertificate     []byte             `yaml:"TLSCertificate,omitempty"`
+	RouterUserConfigs  []RouterUserConfig `yaml:"RouterUserConfigs,omitempty"`
+	AssemblerEndpoints []string           `yaml:"AssemblerEndpoints,omitempty"`
+	TLSCACerts         [][]byte           `yaml:"TLSCACerts,omitempty"`
+	UseTLSRouter       string             `yaml:"UseTLSRouter,omitempty"`
+	UseTLSAssembler    string             `yaml:"UseTLSAssembler,omitempty"`
 }
 
-func NewUserConfig(privateKeyPath string, tlsCertPath string, tlsCACerts [][]byte, network *genconfig.Network) (*UserConfig, error) {
+func NewUserConfig(outputDir string, privateKeyPath string, tlsCertPath string, tlsCACerts [][]byte, network *genconfig.Network) (*UserConfig, error) {
 	// collect router and assembler endpoints, required for defining a user
-	var routerEndpoints []string
+	var routerUserConfigs []RouterUserConfig
 	var assemblerEndpoints []string
 	for _, party := range network.Parties {
-		routerEndpoints = append(routerEndpoints, party.RouterEndpoint)
+		routerUserConfigs = append(routerUserConfigs, RouterUserConfig{
+			Endpoint:   party.RouterEndpoint,
+			PartyID:    party.ID,
+			UserMSPDir: filepath.Join(outputDir, fmt.Sprintf("crypto/ordererOrganizations/org%d/users/user/msp", party.ID)),
+		})
 		assemblerEndpoints = append(assemblerEndpoints, party.AssemblerEndpoint)
 	}
 
@@ -47,7 +59,7 @@ func NewUserConfig(privateKeyPath string, tlsCertPath string, tlsCACerts [][]byt
 	return &UserConfig{
 		TLSPrivateKey:      privateKey,
 		TLSCertificate:     tlsCert,
-		RouterEndpoints:    routerEndpoints,
+		RouterUserConfigs:  routerUserConfigs,
 		AssemblerEndpoints: assemblerEndpoints,
 		TLSCACerts:         tlsCACerts,
 		UseTLSRouter:       network.UseTLSRouter,

test/router_test.go

@@ -16,6 +16,8 @@ import (
 
 	"github.com/hyperledger/fabric-x-orderer/common/tools/armageddon"
 	"github.com/hyperledger/fabric-x-orderer/common/types"
+	"github.com/hyperledger/fabric-x-orderer/common/utils"
+	config "github.com/hyperledger/fabric-x-orderer/config"
 	"github.com/hyperledger/fabric-x-orderer/testutil"
 	"github.com/hyperledger/fabric-x-orderer/testutil/client"
 	"github.com/hyperledger/fabric-x-orderer/testutil/tx"
@@ -350,3 +352,81 @@ func TestSubmitToRouterGetMetrics(t *testing.T) {
 		return testutil.RouterIncomingTxMetric(t, types.PartyID(1), url) == totalTxNumber
 	}, 30*time.Second, 100*time.Millisecond)
 }
+
+// TestVerifySignedTxsByRouterSingleParty verifies that a router running in a single-party,
+// single-shard configuration correctly enforces client signature verification when accepting
+// and broadcasting transactions.
+func TestVerifySignedTxsByRouterSingleParty(t *testing.T) {
+	// 1. compile arma
+	armaBinaryPath, err := gexec.BuildWithEnvironment("github.com/hyperledger/fabric-x-orderer/cmd/arma", []string{"GOPRIVATE=" + os.Getenv("GOPRIVATE")})
+	defer gexec.CleanupBuildArtifacts()
+	require.NoError(t, err)
+	require.NotNil(t, armaBinaryPath)
+
+	t.Logf("Running test with %d parties and %d shards", 1, 1)
+
+	// 2. Create a temporary directory for the test.
+	dir, err := os.MkdirTemp("", t.Name())
+	require.NoError(t, err)
+	defer os.RemoveAll(dir)
+
+	// 3. Create a config YAML file in the temporary directory.
+	configPath := filepath.Join(dir, "config.yaml")
+	netInfo := testutil.CreateNetwork(t, configPath, 1, 1, "none", "none")
+	require.NoError(t, err)
+	numOfArmaNodes := len(netInfo)
+
+	// 4. Generate the config files in the temporary directory using the armageddon generate command.
+	armageddon.NewCLI().Run([]string{"generate", "--config", configPath, "--output", dir})
+
+	configFilePath := filepath.Join(dir, fmt.Sprintf("config/party%d/local_config_router.yaml", types.PartyID(1)))
+	conf, _, err := config.LoadLocalConfig(configFilePath)
+	require.NoError(t, err)
+
+	// Modify the router configuration to require client signature verification.
+	conf.NodeLocalConfig.GeneralConfig.ClientSignatureVerificationRequired = true
+	utils.WriteToYAML(conf.NodeLocalConfig, configFilePath)
+
+	conf, _, err = config.LoadLocalConfig(configFilePath)
+	require.NoError(t, err)
+	require.True(t, conf.NodeLocalConfig.GeneralConfig.ClientSignatureVerificationRequired)
+
+	// 5. Run the arma nodes.
+	// NOTE: if one of the nodes is not started within 10 seconds, there is no point in continuing the test, so fail it
+	// Obtains a test user configuration and constructs a broadcast client.
+	readyChan := make(chan struct{}, numOfArmaNodes)
+	armaNetwork := testutil.RunArmaNodes(t, dir, armaBinaryPath, readyChan, netInfo)
+	defer armaNetwork.Stop()
+
+	testutil.WaitReady(t, readyChan, numOfArmaNodes, 10)
+
+	uc, err := testutil.GetUserConfig(dir, 1)
+	assert.NoError(t, err)
+	assert.NotNil(t, uc)
+
+	totalTxNumber := 1000
+	// rate limiter parameters
+	fillInterval := 10 * time.Millisecond
+	fillFrequency := 1000 / int(fillInterval.Milliseconds())
+	rate := 500
+
+	capacity := rate / fillFrequency
+	rl, err := armageddon.NewRateLimiter(rate, fillInterval, capacity)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "failed to start a rate limiter")
+		os.Exit(3)
+	}
+
+	broadcastClient := client.NewBroadcastTxClient(uc, 10*time.Second)
+
+	for i := range totalTxNumber {
+		status := rl.GetToken()
+		if !status {
+			fmt.Fprintf(os.Stderr, "failed to send tx %d", i+1)
+			os.Exit(3)
+		}
+		txContent := tx.PrepareTxWithTimestamp(i, 64, []byte("sessionNumber"))
+		err = broadcastClient.SendTx(txContent)
+		require.NoError(t, err)
+	}
+}

test/tx_client_test.go

@@ -77,14 +77,14 @@ func TestTxClientSend(t *testing.T) {
 	handler := func(block *common.Block) error {
 		totalTxs += len(block.Data.Data)
 		totalBlocks++
-		if totalTxs == totalTxNumber+1 {
+		if totalBlocks >= int(endBlock-startBlock)+1 {
 			cancel()
 			return context.Canceled
 		}
 		return nil
 	}
 	dc.PullBlocks(cnx, 1, startBlock, endBlock, handler)
-	assert.Equal(t, totalTxNumber+1, totalTxs)
+	assert.GreaterOrEqual(t, totalBlocks, int(endBlock-startBlock)+1)
 	assert.True(t, totalBlocks > 2)
 	t.Logf("Finished pull and count: %d, %d", totalBlocks, totalTxs)
 }

testutil/client/tx_broadcast_client.go

@@ -10,6 +10,8 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"os"
+	"path/filepath"
 	"strings"
 	"sync"
 	"time"
@@ -18,7 +20,9 @@ import (
 	"github.com/hyperledger/fabric-protos-go-apiv2/common"
 	ab "github.com/hyperledger/fabric-protos-go-apiv2/orderer"
 	"github.com/hyperledger/fabric-x-orderer/common/tools/armageddon"
+	"github.com/hyperledger/fabric-x-orderer/common/types"
 	"github.com/hyperledger/fabric-x-orderer/node/comm"
+	"github.com/hyperledger/fabric-x-orderer/node/crypto"
 	"github.com/hyperledger/fabric-x-orderer/testutil/tx"
 )
 
@@ -27,6 +31,9 @@ type StreamInfo struct {
 	totalTxSent uint32
 	endpoint    string
 	streamLock  sync.Mutex
+	partyID     types.PartyID
+	signer      *crypto.ECDSASigner
+	certBytes   []byte
 }
 
 var logger = flogging.MustGetLogger("BroadcastClient")
@@ -42,7 +49,7 @@ func NewBroadcastTxClient(userConfigFile *armageddon.UserConfig, timeOut time.Du
 	return &BroadcastTxClient{
 		userConfig:       userConfigFile,
 		timeOut:          timeOut,
-		streamRoutersMap: make(map[string]*StreamInfo, len(userConfigFile.RouterEndpoints)),
+		streamRoutersMap: make(map[string]*StreamInfo, len(userConfigFile.RouterUserConfigs)),
 	}
 }
 
@@ -78,7 +85,7 @@ func (c *BroadcastTxClient) SendTx(txContent []byte) error {
 			streamInfo.streamLock.Lock()
 			defer streamInfo.streamLock.Unlock()
 
-			env := tx.CreateStructuredEnvelope(txContent)
+			env := tx.CreateSignedStructuredEnvelope(txContent, streamInfo.signer, streamInfo.certBytes, fmt.Sprintf("org%d", streamInfo.partyID))
 			err := streamInfo.stream.Send(env)
 			if err != nil {
 				updateStateLock.Lock()
@@ -115,9 +122,9 @@ func (c *BroadcastTxClient) SendTx(txContent []byte) error {
 
 	waitForReceiveDone.Wait()
 	// check if we had any failures
-	possibleNumOfFailures := len(c.userConfig.RouterEndpoints)
-	if len(c.userConfig.RouterEndpoints) >= 3 {
-		possibleNumOfFailures = len(c.userConfig.RouterEndpoints) / 3
+	possibleNumOfFailures := len(c.userConfig.RouterUserConfigs)
+	if len(c.userConfig.RouterUserConfigs) >= 3 {
+		possibleNumOfFailures = len(c.userConfig.RouterUserConfigs) / 3
 	}
 	if failures > possibleNumOfFailures {
 		er := fmt.Sprintf("\nfailed to send tx to %d out of %d send streams", failures, len(c.streamRoutersMap))
@@ -165,15 +172,20 @@ func (c *BroadcastTxClient) createSendStreams() error {
 	}
 
 	// create gRPC clients and streams to the routers
-	for i := 0; i < len(userConfig.RouterEndpoints); i++ {
-		routerEndpoint := userConfig.RouterEndpoints[i]
+	// routerEnpoints are sorted in the userConfig in the same order as party IDs
+	for _, routerUserConfig := range userConfig.RouterUserConfigs {
+		routerEndpoint := routerUserConfig.Endpoint
 		streamInfo, ok := c.streamRoutersMap[routerEndpoint]
 		if !ok {
 			// create a gRPC connection to the router
 			stream := createSendStream(userConfig, serverRootCAs, routerEndpoint)
 			// if stream is created successfully, add it to the map
 			if stream != nil {
-				c.streamRoutersMap[routerEndpoint] = &StreamInfo{stream: stream, totalTxSent: 0, endpoint: routerEndpoint}
+				signer, certBytes, err := buildCryptoMaterials(routerUserConfig.UserMSPDir)
+				if err != nil {
+					return fmt.Errorf("failed to build crypto materials: %v", err)
+				}
+				c.streamRoutersMap[routerEndpoint] = &StreamInfo{stream: stream, totalTxSent: 0, endpoint: routerEndpoint, partyID: routerUserConfig.PartyID, signer: signer, certBytes: certBytes}
 			}
 		} else {
 			if streamInfo.stream == nil {
@@ -192,6 +204,24 @@ func (c *BroadcastTxClient) createSendStreams() error {
 	return nil
 }
 
+func buildCryptoMaterials(configDir string) (*crypto.ECDSASigner, []byte, error) {
+	keyBytes, err := os.ReadFile(filepath.Join(configDir, "keystore/priv_sk"))
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to read private key file: %v", err)
+	}
+	signer, err := tx.CreateECDSASigner(keyBytes)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to create signer from private key: %v", err)
+	}
+
+	certBytes, err := os.ReadFile(filepath.Join(configDir, "signcerts/sign-cert.pem"))
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to read sign certificate file: %v", err)
+	}
+
+	return signer, certBytes, nil
+}
+
 func (c *BroadcastTxClient) Stop() {
 	c.streamsMapLock.Lock()
 	defer c.streamsMapLock.Unlock()

testutil/tx/tx_utils.go

@@ -17,6 +17,7 @@ import (
 	"time"
 
 	"github.com/hyperledger/fabric-protos-go-apiv2/common"
+	"github.com/hyperledger/fabric-protos-go-apiv2/msp"
 	"github.com/hyperledger/fabric-x-common/protoutil"
 	"github.com/hyperledger/fabric-x-orderer/node/crypto"
 	protos "github.com/hyperledger/fabric-x-orderer/node/protos/comm"
@@ -209,3 +210,34 @@ func PrepareEnvWithTimestamp(txNumber int, envSize int, sessionNumber []byte) *c
 	data := PrepareTxWithTimestamp(txNumber, dataSize, sessionNumber)
 	return CreateStructuredEnvelope(data)
 }
+
+func CreateSignedStructuredEnvelope(data []byte, signer *crypto.ECDSASigner, certBytes []byte, org string) *common.Envelope {
+	payload := createSignedStructuredPayload(data, certBytes, org)
+	payloadBytes := deterministicMarshall(payload)
+	signature, err := signer.Sign(payloadBytes)
+	if err != nil {
+		return nil
+	}
+	return &common.Envelope{
+		Payload:   payloadBytes,
+		Signature: signature,
+	}
+}
+
+func createSignedStructuredPayload(data []byte, certBytes []byte, org string) *common.Payload {
+	payloadChannelHeader := createChannelHeader(common.HeaderType_MESSAGE, 0, "channelID", 0)
+
+	sId := msp.SerializedIdentity{
+		Mspid:   org,
+		IdBytes: certBytes,
+	}
+
+	payloadSignatureHeader := &common.SignatureHeader{
+		Creator: deterministicMarshall(&sId),
+		Nonce:   []byte("nonce"),
+	}
+	return &common.Payload{
+		Header: createPayloadHeader(payloadChannelHeader, payloadSignatureHeader),
+		Data:   data,
+	}
+}