Merge branch 'master' of https://github.com/multiversx/mx-chain-go into rc/supernova

# Conflicts:
#	consensus/spos/bls/v2/errors.go
#	consensus/spos/bls/v2/subroundEndRound.go

Author: sstanculeanu

consensus/spos/bls/proxy/subroundsHandler_test.go

@@ -81,6 +81,7 @@ func getDefaultArgumentsSubroundHandler() (*SubroundsHandlerArgs, *spos.Consensu
 	consensusCore.SetScheduledProcessor(&consensus.ScheduledProcessorStub{})
 	consensusCore.SetMessageSigningHandler(&mock2.MessageSigningHandlerStub{})
 	consensusCore.SetPeerBlacklistHandler(&mock2.PeerBlacklistHandlerStub{})
+	consensusCore.SetPeerSignatureHandler(&cryptoMocks.PeerSignatureHandlerStub{})
 	consensusCore.SetSigningHandler(&consensus.SigningHandlerStub{})
 	consensusCore.SetEnableEpochsHandler(epochsEnable)
 	consensusCore.SetEnableRoundsHandler(&testscommon.EnableRoundsHandlerStub{})

consensus/spos/bls/v2/common.go

@@ -0,0 +1,27 @@
+package v2
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/multiversx/mx-chain-core-go/core"
+
+	"github.com/multiversx/mx-chain-go/consensus/spos"
+)
+
+func checkGoRoutinesThrottler(ctx context.Context, throttler core.Throttler) error {
+	for {
+		if throttler.CanProcess() {
+			break
+		}
+
+		select {
+		case <-time.After(time.Millisecond):
+			continue
+		case <-ctx.Done():
+			return fmt.Errorf("%w while checking the throttler", spos.ErrTimeIsOut)
+		}
+	}
+	return nil
+}

consensus/spos/bls/v2/common_test.go

@@ -0,0 +1,90 @@
+package v2
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	dataRetrieverMock "github.com/multiversx/mx-chain-go/dataRetriever/mock"
+)
+
+func TestCheckGoRoutinesThrottler(t *testing.T) {
+	t.Parallel()
+
+	t.Run("throttler can process should return nil immediately", func(t *testing.T) {
+		t.Parallel()
+
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+
+		throttler := &dataRetrieverMock.ThrottlerStub{
+			CanProcessCalled: func() bool {
+				return true
+			},
+		}
+
+		err := checkGoRoutinesThrottler(ctx, throttler)
+		assert.Nil(t, err)
+	})
+
+	t.Run("throttler cannot process for a period of time, then can process, should return nil", func(t *testing.T) {
+		t.Parallel()
+
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+
+		var numCalls int32
+		throttler := &dataRetrieverMock.ThrottlerStub{
+			CanProcessCalled: func() bool {
+				return atomic.AddInt32(&numCalls, 1) >= 5
+			},
+		}
+
+		err := checkGoRoutinesThrottler(ctx, throttler)
+		assert.Nil(t, err)
+		assert.GreaterOrEqual(t, atomic.LoadInt32(&numCalls), int32(5))
+	})
+
+	t.Run("throttler cannot process should return error when context is done", func(t *testing.T) {
+		t.Parallel()
+
+		ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
+		defer cancel()
+
+		var numCalls int32
+		throttler := &dataRetrieverMock.ThrottlerStub{
+			CanProcessCalled: func() bool {
+				atomic.AddInt32(&numCalls, 1)
+				return false
+			},
+		}
+
+		err := checkGoRoutinesThrottler(ctx, throttler)
+		assert.NotNil(t, err)
+		assert.ErrorContains(t, err, "time is out")
+		assert.ErrorContains(t, err, "while checking the throttler")
+		assert.Greater(t, atomic.LoadInt32(&numCalls), int32(1))
+	})
+
+	t.Run("context already canceled and throttler cannot process should return error without retrying", func(t *testing.T) {
+		t.Parallel()
+
+		ctx, cancel := context.WithCancel(context.Background())
+		cancel()
+
+		var numCalls int32
+		throttler := &dataRetrieverMock.ThrottlerStub{
+			CanProcessCalled: func() bool {
+				atomic.AddInt32(&numCalls, 1)
+				return false
+			},
+		}
+
+		err := checkGoRoutinesThrottler(ctx, throttler)
+		assert.NotNil(t, err)
+		assert.ErrorContains(t, err, "time is out")
+	})
+}

consensus/spos/bls/v2/errors.go

@@ -17,6 +17,12 @@ var ErrValidSignatureFromInvalidSigner = errors.New("valid signature from invali
 // ErrHeaderHashMismatch signals that header hash does not match
 var ErrHeaderHashMismatch = errors.New("header hash does not match")
 
+// ErrPublicKeyMismatch signals that the BLS public key in the message does not belong to the transport sender
+var ErrPublicKeyMismatch = errors.New("public key does not match the message sender")
+
+// ErrSignerNotInConsensusGroup signals that the signer is not part of the current consensus group
+var ErrSignerNotInConsensusGroup = errors.New("signer is not part of the consensus group")
+
 // ErrNilRoundSyncController signals that a nil round sync controller has been provided
 var ErrNilRoundSyncController = errors.New("nil round sync controller")
 

consensus/spos/bls/v2/subroundEndRound.go

@@ -34,6 +34,7 @@ type subroundEndRound struct {
 	mutProcessingEndRound         sync.Mutex
 	sentSignatureTracker          spos.SentSignaturesTracker
 	worker                        spos.WorkerHandler
+	signatureThrottler            core.Throttler
 }
 
 // NewSubroundEndRound creates a subroundEndRound object
@@ -43,6 +44,7 @@ func NewSubroundEndRound(
 	appStatusHandler core.AppStatusHandler,
 	sentSignatureTracker spos.SentSignaturesTracker,
 	worker spos.WorkerHandler,
+	signatureThrottler core.Throttler,
 ) (*subroundEndRound, error) {
 	err := checkNewSubroundEndRoundParams(baseSubround)
 	if err != nil {
@@ -57,6 +59,9 @@ func NewSubroundEndRound(
 	if check.IfNil(worker) {
 		return nil, spos.ErrNilWorker
 	}
+	if check.IfNil(signatureThrottler) {
+		return nil, spos.ErrNilThrottler
+	}
 
 	srEndRound := subroundEndRound{
 		Subround:                      baseSubround,
@@ -65,6 +70,7 @@ func NewSubroundEndRound(
 		mutProcessingEndRound:         sync.Mutex{},
 		sentSignatureTracker:          sentSignatureTracker,
 		worker:                        worker,
+		signatureThrottler:            signatureThrottler,
 	}
 	srEndRound.Job = srEndRound.doEndRoundJob
 	srEndRound.Check = srEndRound.doEndRoundConsensusCheck
@@ -224,11 +230,20 @@ func (sr *subroundEndRound) verifyInvalidSigner(
 		return "", spos.ErrInvalidMessageType
 	}
 
+	if !sr.IsNodeInConsensusGroup(string(cnsMsg.PubKey)) {
+		return "", ErrSignerNotInConsensusGroup
+	}
+
 	err = sr.MessageSigningHandler().Verify(msg)
 	if err != nil {
 		return "", err
 	}
 
+	err = sr.PeerSignatureHandler().VerifyPeerSignature(cnsMsg.PubKey, msg.Peer(), cnsMsg.Signature)
+	if err != nil {
+		return "", ErrPublicKeyMismatch
+	}
+
 	if !bytes.Equal(headerHash, cnsMsg.BlockHeaderHash) {
 		return "", ErrHeaderHashMismatch
 	}
@@ -476,6 +491,10 @@ func (sr *subroundEndRound) aggregateSigsAndHandleInvalidSigners(bitmap []byte,
 	return bitmap, sig, nil
 }
 
+func (sr *subroundEndRound) checkGoRoutinesThrottler(ctx context.Context) error {
+	return checkGoRoutinesThrottler(ctx, sr.signatureThrottler)
+}
+
 // verifySignature implements parallel signature verification
 func (sr *subroundEndRound) verifySignature(i int, pk string, sigShare []byte) error {
 	err := sr.SigningHandler().VerifySignatureShare(uint16(i), sigShare, sr.GetData(), sr.GetHeader().GetEpoch())

consensus/spos/bls/v2/subroundEndRound_test.go

@@ -34,6 +34,7 @@ import (
 	"github.com/multiversx/mx-chain-go/testscommon"
 	consensusMocks "github.com/multiversx/mx-chain-go/testscommon/consensus"
 	"github.com/multiversx/mx-chain-go/testscommon/consensus/initializers"
+	"github.com/multiversx/mx-chain-go/testscommon/cryptoMocks"
 	"github.com/multiversx/mx-chain-go/testscommon/dataRetriever"
 	"github.com/multiversx/mx-chain-go/testscommon/enableEpochsHandlerMock"
 	"github.com/multiversx/mx-chain-go/testscommon/p2pmocks"
@@ -1898,6 +1899,43 @@ func TestVerifyInvalidSigners(t *testing.T) {
 		require.Equal(t, expectedErr, err)
 	})
 
+	t.Run("peer signature binding fails should error", func(t *testing.T) {
+		t.Parallel()
+
+		container := consensusMocks.InitConsensusCore()
+
+		consensusMsg := &consensus.Message{
+			BlockHeaderHash: headerHash,
+			PubKey:          pubKey,
+			MsgType:         int64(bls.MtSignature),
+		}
+		consensusMsgBytes, _ := container.Marshalizer().Marshal(consensusMsg)
+
+		invalidSigners := []p2p.MessageP2P{&factory.Message{
+			FromField: []byte("attackerPid"),
+			DataField: consensusMsgBytes,
+		}}
+		invalidSignersBytes, _ := container.Marshalizer().Marshal(invalidSigners)
+
+		messageSigningHandler := &mock.MessageSigningHandlerStub{
+			DeserializeCalled: func(messagesBytes []byte) ([]p2p.MessageP2P, error) {
+				return invalidSigners, nil
+			},
+		}
+		container.SetMessageSigningHandler(messageSigningHandler)
+
+		container.SetPeerSignatureHandler(&cryptoMocks.PeerSignatureHandlerStub{
+			VerifyPeerSignatureCalled: func(pk []byte, pid core.PeerID, signature []byte) error {
+				return expectedErr
+			},
+		})
+
+		sr := initSubroundEndRoundWithContainer(container, &statusHandler.AppStatusHandlerStub{})
+
+		_, err := sr.VerifyInvalidSigners(headerHash, invalidSignersBytes)
+		require.Equal(t, v2.ErrPublicKeyMismatch, err)
+	})
+
 	t.Run("invalid message type should err", func(t *testing.T) {
 		t.Parallel()
 
@@ -1940,6 +1978,37 @@ func TestVerifyInvalidSigners(t *testing.T) {
 		require.False(t, wasCalled)
 	})
 
+	t.Run("signer not in consensus group should err", func(t *testing.T) {
+		t.Parallel()
+
+		container := consensusMocks.InitConsensusCore()
+
+		unknownPubKey := []byte("unknownKey")
+		consensusMsg := &consensus.Message{
+			PubKey:  unknownPubKey,
+			MsgType: int64(bls.MtSignature),
+		}
+		consensusMsgBytes, _ := container.Marshalizer().Marshal(consensusMsg)
+
+		invalidSigners := []p2p.MessageP2P{&factory.Message{
+			FromField: []byte("from"),
+			DataField: consensusMsgBytes,
+		}}
+		invalidSignersBytes, _ := container.Marshalizer().Marshal(invalidSigners)
+
+		messageSigningHandler := &mock.MessageSigningHandlerStub{
+			DeserializeCalled: func(messagesBytes []byte) ([]p2p.MessageP2P, error) {
+				return invalidSigners, nil
+			},
+		}
+		container.SetMessageSigningHandler(messageSigningHandler)
+
+		sr := initSubroundEndRoundWithContainer(container, &statusHandler.AppStatusHandlerStub{})
+
+		_, err := sr.VerifyInvalidSigners(headerHash, invalidSignersBytes)
+		require.Equal(t, v2.ErrSignerNotInConsensusGroup, err)
+	})
+
 	t.Run("failed to verify signature share", func(t *testing.T) {
 		t.Parallel()
 

consensus/spos/bls/v2/subroundSignature.go

@@ -3,7 +3,6 @@ package v2
 import (
 	"bytes"
 	"context"
-	"fmt"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -290,18 +289,7 @@ func (sr *subroundSignature) sendSignatureForManagedKey(_ context.Context, idx i
 }
 
 func (sr *subroundSignature) checkGoRoutinesThrottler(ctx context.Context) error {
-	for {
-		if sr.signatureThrottler.CanProcess() {
-			break
-		}
-		select {
-		case <-time.After(timeSpentBetweenChecks):
-			continue
-		case <-ctx.Done():
-			return fmt.Errorf("%w while checking the throttler", spos.ErrTimeIsOut)
-		}
-	}
-	return nil
+	return checkGoRoutinesThrottler(ctx, sr.signatureThrottler)
 }
 
 func (sr *subroundSignature) doSignatureJobForSingleKey(_ context.Context) bool {

consensus/spos/consensusCore.go

@@ -4,6 +4,7 @@ import (
 	"github.com/multiversx/mx-chain-core-go/data"
 	"github.com/multiversx/mx-chain-core-go/hashing"
 	"github.com/multiversx/mx-chain-core-go/marshal"
+	crypto "github.com/multiversx/mx-chain-crypto-go"
 
 	"github.com/multiversx/mx-chain-go/common"
 	cryptoCommon "github.com/multiversx/mx-chain-go/common/crypto"
@@ -40,6 +41,7 @@ type ConsensusCore struct {
 	scheduledProcessor            consensus.ScheduledProcessor
 	messageSigningHandler         consensus.P2PSigningHandler
 	peerBlacklistHandler          consensus.PeerBlacklistHandler
+	peerSignatureHandler          crypto.PeerSignatureHandler
 	signingHandler                consensus.SigningHandler
 	enableEpochsHandler           common.EnableEpochsHandler
 	enableRoundsHandler           common.EnableRoundsHandler
@@ -74,6 +76,7 @@ type ConsensusCoreArgs struct {
 	ScheduledProcessor            consensus.ScheduledProcessor
 	MessageSigningHandler         consensus.P2PSigningHandler
 	PeerBlacklistHandler          consensus.PeerBlacklistHandler
+	PeerSignatureHandler          crypto.PeerSignatureHandler
 	SigningHandler                consensus.SigningHandler
 	EnableEpochsHandler           common.EnableEpochsHandler
 	EnableRoundsHandler           common.EnableRoundsHandler
@@ -111,6 +114,7 @@ func NewConsensusCore(
 		scheduledProcessor:            args.ScheduledProcessor,
 		messageSigningHandler:         args.MessageSigningHandler,
 		peerBlacklistHandler:          args.PeerBlacklistHandler,
+		peerSignatureHandler:          args.PeerSignatureHandler,
 		signingHandler:                args.SigningHandler,
 		enableEpochsHandler:           args.EnableEpochsHandler,
 		enableRoundsHandler:           args.EnableRoundsHandler,
@@ -244,6 +248,11 @@ func (cc *ConsensusCore) PeerBlacklistHandler() consensus.PeerBlacklistHandler {
 	return cc.peerBlacklistHandler
 }
 
+// PeerSignatureHandler will return the peer signature handler
+func (cc *ConsensusCore) PeerSignatureHandler() crypto.PeerSignatureHandler {
+	return cc.peerSignatureHandler
+}
+
 // SigningHandler will return the signing handler component
 func (cc *ConsensusCore) SigningHandler() consensus.SigningHandler {
 	return cc.signingHandler
@@ -364,6 +373,11 @@ func (cc *ConsensusCore) SetPeerBlacklistHandler(peerBlacklistHandler consensus.
 	cc.peerBlacklistHandler = peerBlacklistHandler
 }
 
+// SetPeerSignatureHandler sets peer signature handler
+func (cc *ConsensusCore) SetPeerSignatureHandler(peerSignatureHandler crypto.PeerSignatureHandler) {
+	cc.peerSignatureHandler = peerSignatureHandler
+}
+
 // SetHeaderSigVerifier sets header sig verifier
 func (cc *ConsensusCore) SetHeaderSigVerifier(headerSigVerifier consensus.HeaderSigVerifier) {
 	cc.headerSigVerifier = headerSigVerifier

consensus/spos/consensusCoreValidator.go

@@ -76,6 +76,9 @@ func ValidateConsensusCore(container ConsensusCoreHandler) error {
 	if check.IfNil(container.PeerBlacklistHandler()) {
 		return ErrNilPeerBlacklistHandler
 	}
+	if check.IfNil(container.PeerSignatureHandler()) {
+		return ErrNilPeerSignatureHandler
+	}
 	if check.IfNil(container.SigningHandler()) {
 		return ErrNilSigningHandler
 	}