fix after review: small refactor, in order to keep the mutex locked only when necessary + added equivalent message signature verification

Author: sstanculeanu

consensus/spos/worker.go

@@ -721,14 +721,7 @@ func (wrk *Worker) ResetConsensusMessages() {
 }
 
 func (wrk *Worker) checkValidityAndProcessEquivalentMessages(cnsMsg *consensus.Message, p2pMessage p2p.MessageP2P) error {
-	wrk.mutEquivalentMessages.Lock()
-	defer wrk.mutEquivalentMessages.Unlock()
-
 	msgType := consensus.MessageType(cnsMsg.MsgType)
-	err := wrk.processEquivalentMessageUnprotected(msgType, cnsMsg.BlockHeaderHash)
-	if err != nil {
-		return err
-	}
 
 	log.Trace("received message from consensus topic",
 		"msg type", wrk.consensusService.GetStringValue(msgType),
@@ -738,28 +731,41 @@ func (wrk *Worker) checkValidityAndProcessEquivalentMessages(cnsMsg *consensus.M
 		"size", len(p2pMessage.Data()),
 	)
 
+	if !wrk.enableEpochsHandler.IsEquivalentMessagesFlagEnabled() {
+		return wrk.consensusMessageValidator.checkConsensusMessageValidity(cnsMsg, p2pMessage.Peer())
+	}
+
+	// if the message is not with final info, no need to check its equivalent messages
+	if !wrk.consensusService.IsMessageWithFinalInfo(msgType) {
+		return wrk.consensusMessageValidator.checkConsensusMessageValidity(cnsMsg, p2pMessage.Peer())
+	}
+
+	wrk.mutEquivalentMessages.Lock()
+	defer wrk.mutEquivalentMessages.Unlock()
+
+	err := wrk.processEquivalentMessageUnprotected(cnsMsg)
+	if err != nil {
+		return err
+	}
+
 	err = wrk.consensusMessageValidator.checkConsensusMessageValidity(cnsMsg, p2pMessage.Peer())
 	if err != nil {
-		wrk.processInvalidEquivalentMessageUnprotected(msgType, cnsMsg.BlockHeaderHash)
+		wrk.processInvalidEquivalentMessageUnprotected(cnsMsg.BlockHeaderHash)
 		return err
 	}
 
 	return nil
 }
 
-func (wrk *Worker) processEquivalentMessageUnprotected(msgType consensus.MessageType, blockHeaderHash []byte) error {
-	if wrk.enableEpochsHandler.IsEquivalentMessagesFlagEnabled() {
-		return nil
-	}
-
-	// early exit if the message is not with final info
-	if !wrk.consensusService.IsMessageWithFinalInfo(msgType) {
-		return nil
+func (wrk *Worker) processEquivalentMessageUnprotected(cnsMsg *consensus.Message) error {
+	err := wrk.verifyEquivalentMessageSignature(cnsMsg)
+	if err != nil {
+		return err
 	}
 
-	hdrHash := string(blockHeaderHash)
+	hdrHash := string(cnsMsg.BlockHeaderHash)
 
-	// if an equivalent message was seen before, return error to stop further broadcasts
+	// if a valid equivalent message was seen before, return error to stop further broadcasts
 	numMessages := wrk.equivalentMessages[hdrHash]
 	wrk.equivalentMessages[hdrHash] = numMessages + 1
 	if numMessages > 0 {
@@ -769,17 +775,34 @@ func (wrk *Worker) processEquivalentMessageUnprotected(msgType consensus.Message
 	return nil
 }
 
-func (wrk *Worker) processInvalidEquivalentMessageUnprotected(msgType consensus.MessageType, blockHeaderHash []byte) {
-	if wrk.enableEpochsHandler.IsEquivalentMessagesFlagEnabled() {
-		return
+func (wrk *Worker) verifyEquivalentMessageSignature(cnsMsg *consensus.Message) error {
+	if check.IfNil(wrk.consensusState.Header) {
+		return ErrNilHeader
 	}
 
-	if !wrk.consensusService.IsMessageWithFinalInfo(msgType) {
-		return
+	header := wrk.consensusState.Header.ShallowClone()
+
+	err := header.SetSignature(cnsMsg.Signature)
+	if err != nil {
+		return err
 	}
 
-	hdrHash := string(blockHeaderHash)
+	err = header.SetPubKeysBitmap(cnsMsg.PubKeysBitmap)
+	if err != nil {
+		return err
+	}
 
+	err = wrk.headerSigVerifier.VerifySignature(header)
+	if err != nil {
+		log.Debug("verifyEquivalentMessageSignature", "error", err.Error())
+		return err
+	}
+
+	return nil
+}
+
+func (wrk *Worker) processInvalidEquivalentMessageUnprotected(blockHeaderHash []byte) {
+	hdrHash := string(blockHeaderHash)
 	delete(wrk.equivalentMessages, hdrHash)
 }
 

consensus/spos/worker_test.go

@@ -608,10 +608,13 @@ func TestWorker_ProcessReceivedMessageRedundancyNodeShouldResetInactivityIfNeede
 	assert.True(t, wasCalled)
 }
 
-func TestWorker_ProcessReceivedMessageEquivalentMessageShouldReturnError(t *testing.T) {
+func TestWorker_ProcessReceivedMessageEquivalentMessage(t *testing.T) {
 	t.Parallel()
 
 	workerArgs := createDefaultWorkerArgs(&statusHandlerMock.AppStatusHandlerStub{})
+	workerArgs.EnableEpochsHandler = &enableEpochsHandlerMock.EnableEpochsHandlerStub{
+		IsEquivalentMessagesFlagEnabledField: true,
+	}
 	wrk, _ := spos.NewWorker(workerArgs)
 
 	equivalentBlockHeaderHash := workerArgs.Hasher.Compute("equivalent block header hash")
@@ -685,6 +688,27 @@ func TestWorker_ProcessReceivedMessageEquivalentMessageShouldReturnError(t *test
 		fromConnectedPeerId,
 		&p2pmocks.MessengerStub{},
 	)
+	assert.Equal(t, spos.ErrNilHeader, err)
+
+	wrk.ConsensusState().Header = &block.Header{
+		ChainID:         chainID,
+		PrevHash:        []byte("prev hash"),
+		PrevRandSeed:    []byte("prev rand seed"),
+		RandSeed:        []byte("rand seed"),
+		RootHash:        []byte("roothash"),
+		SoftwareVersion: []byte("software version"),
+		AccumulatedFees: big.NewInt(0),
+		DeveloperFees:   big.NewInt(0),
+	}
+	err = wrk.ProcessReceivedMessage(
+		&p2pmocks.P2PMessageMock{
+			DataField:      buff,
+			PeerField:      currentPid,
+			SignatureField: []byte("signature"),
+		},
+		fromConnectedPeerId,
+		&p2pmocks.MessengerStub{},
+	)
 	assert.NoError(t, err)
 
 	equivalentMessages := wrk.GetEquivalentMessages()