Guard emergency recovery ViewID transitions (#5107)

* fix: guard emergency recovery ViewID transitions

* fix: apply recovery ViewID floor to shard 1

Author: Frozen

cmd/harmony/main.go

@@ -783,16 +783,41 @@ func setupConsensusAndNode(hc harmonyconfig.HarmonyConfig, nodeConfig *nodeconfi
 			Msg("InitConsensusWithMembers failed")
 	}
 
-	// Set the consensus ID to be the current block number
+	// Offline maintenance never starts consensus signing, so it must remain
+	// usable for preflight and rollback even before release constants are filled.
+	// Every networked validator installs the floor before any signing service.
+	if !hc.General.IsOffline {
+		if err := currentConsensus.ConfigureEmergencyRecoveryViewIDFloor(); err != nil {
+			utils.Logger().Panic().Err(err).
+				Msg("refusing to start without a valid emergency recovery ViewID floor")
+		}
+	}
+
+	// Set the consensus ID to the checked successor of the retained head's view.
 	viewID := currentNode.Blockchain().CurrentBlock().Header().ViewID().Uint64()
-	currentConsensus.SetViewIDs(viewID + 1)
+	nextViewID, err := consensus.CheckedNextViewID(viewID)
+	if err != nil {
+		utils.Logger().Panic().Err(err).
+			Uint64("headViewID", viewID).
+			Msg("refusing to start because the head ViewID cannot advance")
+	}
+	currentConsensus.SetViewIDs(nextViewID)
 	utils.Logger().Info().
-		Uint64("viewID", viewID).
+		Uint64("headViewID", viewID).
+		Uint64("consensusViewID", currentConsensus.GetCurBlockViewID()).
 		Msg("Init Blockchain")
 
 	currentNode.Consensus.Registry().SetNodeConfig(currentNode.NodeConfig)
-	// update consensus information based on the blockchain
-	currentConsensus.SetMode(currentConsensus.UpdateConsensusInformation("setupConsensusAndNode"))
+	// Update the committee first, then derive the leader from the exact recovery
+	// view gap before StartChannel can trigger a proposal.
+	mode := currentConsensus.UpdateConsensusInformation("setupConsensusAndNode")
+	if !hc.General.IsOffline {
+		if err := currentConsensus.InitializeEmergencyRecoveryLeader(); err != nil {
+			utils.Logger().Panic().Err(err).
+				Msg("refusing to start without a deterministic emergency recovery leader")
+		}
+	}
+	currentConsensus.SetMode(mode)
 	currentConsensus.NextBlockDue = time.Now()
 	return currentNode
 }

consensus/checks.go

@@ -142,6 +142,12 @@ func (consensus *Consensus) newBlockSanityChecks(
 
 // TODO: leo: move the sanity check to p2p message validation
 func (consensus *Consensus) onViewChangeSanityCheck(recvMsg *FBFTMessage) bool {
+	if err := consensus.assertEmergencyRecoveryViewID(recvMsg.ViewID); err != nil {
+		consensus.getLogger().Warn().Err(err).
+			Uint64("MsgViewChangingID", recvMsg.ViewID).
+			Msg("[onViewChangeSanityCheck] rejected ViewID below recovery floor")
+		return false
+	}
 	// TODO: if difference is only one, new leader can still propose the same committed block to avoid another view change
 	// TODO: new leader catchup without ignore view change message
 
@@ -194,6 +200,12 @@ func (consensus *Consensus) onViewChangeSanityCheck(recvMsg *FBFTMessage) bool {
 
 // TODO: leo: move the sanity check to p2p message validation
 func (consensus *Consensus) onNewViewSanityCheck(recvMsg *FBFTMessage) bool {
+	if err := consensus.assertEmergencyRecoveryViewID(recvMsg.ViewID); err != nil {
+		consensus.getLogger().Warn().Err(err).
+			Uint64("MsgViewChangingID", recvMsg.ViewID).
+			Msg("[onNewView] rejected ViewID below recovery floor")
+		return false
+	}
 	if recvMsg.ViewID < consensus.getCurBlockViewID() {
 		consensus.getLogger().Warn().
 			Uint64("LastSuccessfulConsensusViewID", consensus.getCurBlockViewID()).
@@ -203,3 +215,21 @@ func (consensus *Consensus) onNewViewSanityCheck(recvMsg *FBFTMessage) bool {
 	}
 	return true
 }
+
+func (consensus *Consensus) validateExpectedNewViewLeader(sender *bls.PublicKeyWrapper, viewID uint64) error {
+	if sender == nil {
+		return errors.New("NEWVIEW sender is missing")
+	}
+	expected := consensus.getLeaderPubKey()
+	if consensus.current.GetViewIDFloor() > 0 {
+		var err error
+		expected, err = consensus.expectedLeaderForViewID(viewID)
+		if err != nil {
+			return err
+		}
+	}
+	if expected == nil || sender.Bytes != expected.Bytes {
+		return errors.New("NEWVIEW sender is not the selected leader for ViewID")
+	}
+	return nil
+}

consensus/consensus_block_proposing.go

@@ -23,6 +23,9 @@ const (
 
 // ProposeNewBlock proposes a new block...
 func (consensus *Consensus) ProposeNewBlock(now time.Time, commitSigs chan []byte) (*types.Block, error) {
+	if err := consensus.assertEmergencyRecoveryViewID(consensus.getCurBlockViewID()); err != nil {
+		return nil, errors.Wrap(err, "refusing to propose with unsafe ViewID")
+	}
 	var (
 		currentHeader = consensus.Blockchain().CurrentHeader()
 		nowEpoch      = currentHeader.Epoch()

consensus/consensus_service.go

@@ -268,6 +268,9 @@ func (consensus *Consensus) RegisterRndChannel(rndChannel chan [548]byte) {
 
 // Check viewID, caller's responsibility to hold lock when change ignoreViewIDCheck
 func (consensus *Consensus) checkViewID(msg *FBFTMessage) error {
+	if err := consensus.assertEmergencyRecoveryViewID(msg.ViewID); err != nil {
+		return err
+	}
 	// just ignore consensus check for the first time when node join
 	if consensus.IgnoreViewIDCheck.IsSet() {
 		//in syncing mode, node accepts incoming messages without viewID/leaderKey checking
@@ -494,8 +497,9 @@ func (consensus *Consensus) updateConsensusInformation(reason string) Mode {
 			}
 
 			// If the leader changed and I myself become the leader
-			if (oldLeader != nil && consensus.getLeaderPubKey() != nil &&
-				!consensus.getLeaderPubKey().Object.IsEqual(oldLeader.Object)) && consensus.isLeader() {
+			if consensus.current.GetViewIDFloor() == 0 &&
+				(oldLeader != nil && consensus.getLeaderPubKey() != nil &&
+					!consensus.getLeaderPubKey().Object.IsEqual(oldLeader.Object)) && consensus.isLeader() {
 				go func() {
 					consensus.GetLogger().Info().
 						Str("myKey", myPubKeys.SerializeToHexStr()).
@@ -619,6 +623,9 @@ func (consensus *Consensus) selfCommit(payload []byte) error {
 	if block == nil {
 		return errGetPreparedBlock
 	}
+	if err := consensus.assertEmergencyRecoveryBlockViewID(block.Header().ViewID().Uint64()); err != nil {
+		return err
+	}
 
 	aggSig, mask, err := readSignatureBitmapPayload(payload, 32, consensus.decider().Participants())
 	if err != nil {

consensus/consensus_v2.go

@@ -377,8 +377,16 @@ func (consensus *Consensus) StartChannel() {
 func (consensus *Consensus) syncReadyChan(reason string) {
 	consensus.getLogger().Info().Msgf("[ConsensusMainLoop] syncReadyChan %s", reason)
 	if consensus.getBlockNum() < consensus.Blockchain().CurrentHeader().Number().Uint64()+1 {
-		consensus.setBlockNum(consensus.Blockchain().CurrentHeader().Number().Uint64() + 1)
-		consensus.setViewIDs(consensus.Blockchain().CurrentHeader().ViewID().Uint64() + 1)
+		currentHeader := consensus.Blockchain().CurrentHeader()
+		nextViewID, err := checkedNextViewID(currentHeader.ViewID().Uint64())
+		if err != nil {
+			consensus.getLogger().Error().Err(err).
+				Uint64("headViewID", currentHeader.ViewID().Uint64()).
+				Msg("[syncReadyChan] refusing to reset to an exhausted ViewID")
+			return
+		}
+		consensus.setBlockNum(currentHeader.Number().Uint64() + 1)
+		consensus.setViewIDs(nextViewID)
 		mode := consensus.updateConsensusInformation(reason)
 		consensus.current.SetMode(mode)
 		consensus.getLogger().Info().Msg("[syncReadyChan] Start consensus timer")
@@ -560,6 +568,11 @@ func (consensus *Consensus) preCommitAndPropose(blk *types.Block) error {
 	if blk == nil {
 		return errors.New("block to pre-commit is nil")
 	}
+	if consensus.current.GetViewIDFloor() > 0 {
+		if err := consensus.validateEmergencyRecoveryMessageBlockViewID(blk, consensus.getCurBlockViewID()); err != nil {
+			return err
+		}
+	}
 
 	leaderPriKey, err := consensus.getConsensusLeaderPrivateKey()
 	if err != nil {
@@ -579,13 +592,19 @@ func (consensus *Consensus) preCommitAndPropose(blk *types.Block) error {
 		network.Bytes,
 		network.FBFTMsg
 	bareMinimumCommit := FBFTMsg.Payload
-	consensus.fBFTLog.AddVerifiedMessage(FBFTMsg)
 
 	if err := consensus.verifyLastCommitSig(bareMinimumCommit, blk); err != nil {
 		return errors.Wrap(err, "[preCommitAndPropose] failed verifying last commit sig")
 	}
+	consensus.fBFTLog.AddVerifiedMessage(FBFTMsg)
 
 	go func() {
+		if consensus.current.GetViewIDFloor() > 0 {
+			if err := consensus.validateEmergencyRecoveryMessageBlockViewID(blk, FBFTMsg.ViewID); err != nil {
+				consensus.GetLogger().Error().Err(err).Msg("[preCommitAndPropose] unsafe recovery ViewID")
+				return
+			}
+		}
 		blk.SetCurrentCommitSig(bareMinimumCommit)
 
 		// Send committed message to validators since 2/3 commit is already collected
@@ -675,6 +694,13 @@ func (consensus *Consensus) tryCatchup() error {
 		if blk == nil {
 			return nil
 		}
+		if msg == nil {
+			return errors.New("[TryCatchup] committed message is missing")
+		}
+		if err := consensus.validateEmergencyRecoveryMessageBlockViewID(blk, msg.ViewID); err != nil {
+			consensus.getLogger().Error().Err(err).Msg("[TryCatchup] unsafe recovery ViewID")
+			return err
+		}
 		blk.SetCurrentCommitSig(msg.Payload)
 
 		if err := consensus.verifyBlock(blk); err != nil {
@@ -700,6 +726,12 @@ func (consensus *Consensus) tryCatchup() error {
 }
 
 func (consensus *Consensus) commitBlock(blk *types.Block, committedMsg *FBFTMessage) error {
+	if committedMsg == nil {
+		return errors.New("committed message is missing")
+	}
+	if err := consensus.validateEmergencyRecoveryMessageBlockViewID(blk, committedMsg.ViewID); err != nil {
+		return err
+	}
 	if consensus.Blockchain().CurrentBlock().NumberU64() < blk.NumberU64() {
 		_, err := consensus.Blockchain().InsertChain([]*types.Block{blk}, !consensus.fBFTLog.IsBlockVerified(blk.Hash()))
 		if err != nil && !errors.Is(err, core.ErrKnownBlock) {
@@ -715,7 +747,9 @@ func (consensus *Consensus) commitBlock(blk *types.Block, committedMsg *FBFTMess
 
 	consensus.FinishFinalityCount()
 	consensus.postConsensusProcessing(blk)
-	consensus.setupForNewConsensus(blk, committedMsg)
+	if err := consensus.setupForNewConsensus(blk, committedMsg); err != nil {
+		return err
+	}
 	consensus.getLogger().Info().Uint64("blockNum", blk.NumberU64()).
 		Str("hash", blk.Header().Hash().Hex()).
 		Msg("Added New Block to Blockchain!!!")
@@ -840,9 +874,13 @@ func (consensus *Consensus) rotateLeader(epoch *big.Int, defaultKey *bls.PublicK
 }
 
 // SetupForNewConsensus sets the state for new consensus
-func (consensus *Consensus) setupForNewConsensus(blk *types.Block, committedMsg *FBFTMessage) {
+func (consensus *Consensus) setupForNewConsensus(blk *types.Block, committedMsg *FBFTMessage) error {
+	nextViewID, err := checkedNextViewID(committedMsg.ViewID)
+	if err != nil {
+		return errors.Wrap(err, "cannot advance consensus ViewID")
+	}
 	consensus.setBlockNum(blk.NumberU64() + 1)
-	consensus.setCurBlockViewID(committedMsg.ViewID + 1)
+	consensus.setCurBlockViewID(nextViewID)
 	var epoch *big.Int
 	if blk.IsLastBlockInEpoch() {
 		epoch = new(big.Int).Add(blk.Epoch(), common.Big1)
@@ -897,6 +935,7 @@ func (consensus *Consensus) setupForNewConsensus(blk *types.Block, committedMsg
 	consensus.fBFTLog.PruneCacheBeforeBlock(blk.NumberU64())
 	consensus.resetState()
 	consensus.sendLastSignPower()
+	return nil
 }
 
 func (consensus *Consensus) postCatchup(initBN uint64) {

consensus/construct.go

@@ -58,6 +58,15 @@ func (pm *State) populateMessageFieldsAndSender(
 func (consensus *Consensus) construct(
 	p msg_pb.MessageType, payloadForSign []byte, priKeys []*bls.PrivateKeyWrapper,
 ) (*NetworkMessage, error) {
+	if err := consensus.assertEmergencyRecoveryViewID(consensus.getCurBlockViewID()); err != nil {
+		return nil, err
+	}
+	switch p {
+	case msg_pb.MessageType_ANNOUNCE, msg_pb.MessageType_PREPARE, msg_pb.MessageType_PREPARED:
+		if err := consensus.validateCurrentConsensusBlockViewID(); err != nil {
+			return nil, err
+		}
+	}
 	if len(priKeys) == 0 {
 		return nil, errors.New("no elected bls keys provided")
 	}

consensus/leader.go

@@ -18,6 +18,10 @@ import (
 
 // announce fires leader
 func (consensus *Consensus) announce(block *types.Block) {
+	if err := consensus.assertEmergencyRecoveryBlockViewID(block.Header().ViewID().Uint64()); err != nil {
+		consensus.getLogger().Error().Err(err).Msg("[Announce] unsafe recovery ViewID")
+		return
+	}
 	blockHash := block.Hash()
 
 	// prepare message and broadcast to validators
@@ -269,6 +273,12 @@ func (consensus *Consensus) onCommit(recvMsg *FBFTMessage) {
 			Msg("[OnCommit] Failed finding a matching block for committed message")
 		return
 	}
+	if consensus.current.GetViewIDFloor() > 0 {
+		if err := consensus.validateEmergencyRecoveryMessageBlockViewID(blockObj, recvMsg.ViewID); err != nil {
+			consensus.getLogger().Warn().Err(err).Msg("[OnCommit] unsafe recovery ViewID")
+			return
+		}
+	}
 	commitPayload := signature.ConstructCommitPayload(consensus.Blockchain().Config(),
 		blockObj.Epoch(), blockObj.Hash(), blockObj.NumberU64(), blockObj.Header().ViewID().Uint64())
 	logger = logger.With().