diff --git a/node/node_cross_link.go b/node/node_cross_link.go new file mode 100644 index 000000000..df73feb9c --- /dev/null +++ b/node/node_cross_link.go @@ -0,0 +1,185 @@ +package node + +import ( + "encoding/binary" + + "github.com/ethereum/go-ethereum/rlp" + "github.com/harmony-one/bls/ffi/go/bls" + "github.com/harmony-one/harmony/consensus/quorum" + "github.com/harmony-one/harmony/core/types" + bls_cosi "github.com/harmony-one/harmony/crypto/bls" + "github.com/harmony-one/harmony/internal/ctxerror" + "github.com/harmony-one/harmony/internal/utils" + "github.com/harmony-one/harmony/shard" +) + +const ( + maxPendingCrossLinkSize = 300 + crossLinkBatchSize = 10 +) + +// VerifyBlockCrossLinks verifies the cross links of the block +func (node *Node) VerifyBlockCrossLinks(block *types.Block) error { + if len(block.Header().CrossLinks()) == 0 { + utils.Logger().Debug().Msgf("[CrossLinkVerification] Zero CrossLinks in the header") + return nil + } + + crossLinks := &types.CrossLinks{} + err := rlp.DecodeBytes(block.Header().CrossLinks(), crossLinks) + if err != nil { + return ctxerror.New("[CrossLinkVerification] failed to decode cross links", + "blockHash", block.Hash(), + "crossLinks", len(block.Header().CrossLinks()), + ).WithCause(err) + } + + if !crossLinks.IsSorted() { + return ctxerror.New("[CrossLinkVerification] cross links are not sorted", + "blockHash", block.Hash(), + "crossLinks", len(block.Header().CrossLinks()), + ) + } + + for _, crossLink := range *crossLinks { + cl, err := node.Blockchain().ReadCrossLink(crossLink.ShardID(), crossLink.BlockNum()) + if err == nil && cl != nil { + // Add slash for exist same blocknum but different crosslink + return ctxerror.New("crosslink already exist!") + } + if err = node.VerifyCrossLink(crossLink); err != nil { + return ctxerror.New("cannot VerifyBlockCrossLinks", + "blockHash", block.Hash(), + "blockNum", block.Number(), + "crossLinkShard", crossLink.ShardID(), + "crossLinkBlock", crossLink.BlockNum(), + "numTx", len(block.Transactions()), + ).WithCause(err) + } + } + return nil +} + +// ProcessCrossLinkMessage verify and process Node/CrossLink message into crosslink when it's valid +func (node *Node) ProcessCrossLinkMessage(msgPayload []byte) { + if node.NodeConfig.ShardID == 0 { + pendingCLs, err := node.Blockchain().ReadPendingCrossLinks() + if err == nil && len(pendingCLs) >= maxPendingCrossLinkSize { + utils.Logger().Debug(). + Msgf("[ProcessingCrossLink] Pending Crosslink reach maximum size: %d", len(pendingCLs)) + return + } + + var crosslinks []types.CrossLink + err = rlp.DecodeBytes(msgPayload, &crosslinks) + if err != nil { + utils.Logger().Error(). + Err(err). + Msg("[ProcessingCrossLink] Crosslink Message Broadcast Unable to Decode") + return + } + + candidates := []types.CrossLink{} + utils.Logger().Debug(). + Msgf("[ProcessingCrossLink] Received crosslinks: %d", len(crosslinks)) + + for i, cl := range crosslinks { + if i > crossLinkBatchSize { + break + } + exist, err := node.Blockchain().ReadCrossLink(cl.ShardID(), cl.Number().Uint64()) + if err == nil && exist != nil { + // TODO: leader add double sign checking + utils.Logger().Err(err). + Msgf("[ProcessingCrossLink] Cross Link already exists, pass. Block num: %d, shardID %d", cl.Number(), cl.ShardID()) + continue + } + + if err = node.VerifyCrossLink(cl); err != nil { + utils.Logger().Err(err). + Msgf("[ProcessingCrossLink] Failed to verify new cross link for blockNum %d epochNum %d shard %d skipped: %v", cl.BlockNum(), cl.Epoch().Uint64(), cl.ShardID(), cl) + continue + } + + candidates = append(candidates, cl) + utils.Logger().Debug(). + Msgf("[ProcessingCrossLink] Committing for shardID %d, blockNum %d", cl.ShardID(), cl.Number().Uint64()) + } + node.pendingCLMutex.Lock() + Len, _ := node.Blockchain().AddPendingCrossLinks(candidates) + node.pendingCLMutex.Unlock() + utils.Logger().Debug(). + Msgf("[ProcessingCrossLink] Add pending crosslinks, total pending: %d", Len) + } +} + +// VerifyCrossLink verifies the header is valid +func (node *Node) VerifyCrossLink(cl types.CrossLink) error { + if node.Blockchain().ShardID() != shard.BeaconChainShardID { + return ctxerror.New("[VerifyCrossLink] Shard chains should not verify cross links") + } + + if cl.BlockNum() <= 1 { + return ctxerror.New("[VerifyCrossLink] CrossLink BlockNumber should greater than 1") + } + + if !node.Blockchain().Config().IsCrossLink(cl.Epoch()) { + return ctxerror.New("[VerifyCrossLink] CrossLink Epoch should >= cross link starting epoch", "crossLinkEpoch", cl.Epoch(), "cross_link_starting_eoch", node.Blockchain().Config().CrossLinkEpoch) + } + + // Verify signature of the new cross link header + // TODO: check whether to recalculate shard state + shardState, err := node.Blockchain().ReadShardState(cl.Epoch()) + committee := shardState.FindCommitteeByID(cl.ShardID()) + + if err != nil || committee == nil { + return ctxerror.New("[VerifyCrossLink] Failed to read shard state for cross link", "shardID", cl.ShardID(), "blockNum", cl.BlockNum()).WithCause(err) + } + var committerKeys []*bls.PublicKey + + parseKeysSuccess := true + for _, member := range committee.Slots { + committerKey := new(bls.PublicKey) + err = member.BlsPublicKey.ToLibBLSPublicKey(committerKey) + if err != nil { + parseKeysSuccess = false + break + } + committerKeys = append(committerKeys, committerKey) + } + if !parseKeysSuccess { + return ctxerror.New("[VerifyCrossLink] cannot convert BLS public key", "shardID", cl.ShardID(), "blockNum", cl.BlockNum()).WithCause(err) + } + + mask, err := bls_cosi.NewMask(committerKeys, nil) + if err != nil { + return ctxerror.New("[VerifyCrossLink] cannot create group sig mask", "shardID", cl.ShardID(), "blockNum", cl.BlockNum()).WithCause(err) + } + if err := mask.SetMask(cl.Bitmap()); err != nil { + return ctxerror.New("[VerifyCrossLink] cannot set group sig mask bits", "shardID", cl.ShardID(), "blockNum", cl.BlockNum()).WithCause(err) + } + + decider := quorum.NewDecider(quorum.SuperMajorityStake) + if _, err := decider.SetVoters(committee.Slots); err != nil { + return ctxerror.New("[VerifyCrossLink] Cannot SetVoters for committee", "shardID", cl.ShardID()) + } + if !decider.IsQuorumAchievedByMask(mask) { + return ctxerror.New("[VerifyCrossLink] Not enough voting power for crosslink", "shardID", cl.ShardID()) + } + + aggSig := bls.Sign{} + sig := cl.Signature() + err = aggSig.Deserialize(sig[:]) + if err != nil { + return ctxerror.New("[VerifyCrossLink] unable to deserialize multi-signature from payload").WithCause(err) + } + + hash := cl.Hash() + blockNumBytes := make([]byte, 8) + binary.LittleEndian.PutUint64(blockNumBytes, cl.BlockNum()) + commitPayload := append(blockNumBytes, hash[:]...) + if !aggSig.VerifyHash(mask.AggregatePublic, commitPayload) { + return ctxerror.New("[VerifyCrossLink] Failed to verify the signature for cross link", "shardID", cl.ShardID(), "blockNum", cl.BlockNum()) + } + return nil +} diff --git a/node/node_cross_shard.go b/node/node_cross_shard.go index 14fb2256c..350a2da46 100644 --- a/node/node_cross_shard.go +++ b/node/node_cross_shard.go @@ -1,15 +1,11 @@ package node import ( - "encoding/binary" - "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/rlp" - "github.com/harmony-one/bls/ffi/go/bls" proto_node "github.com/harmony-one/harmony/api/proto/node" "github.com/harmony-one/harmony/core" "github.com/harmony-one/harmony/core/types" - bls_cosi "github.com/harmony-one/harmony/crypto/bls" nodeconfig "github.com/harmony-one/harmony/internal/configs/node" "github.com/harmony-one/harmony/internal/ctxerror" "github.com/harmony-one/harmony/internal/utils" @@ -97,90 +93,6 @@ func (node *Node) BroadcastMissingCXReceipts() { } } -// VerifyBlockCrossLinks verifies the cross links of the block -func (node *Node) VerifyBlockCrossLinks(block *types.Block) error { - if len(block.Header().CrossLinks()) == 0 { - utils.Logger().Debug().Msgf("[CrossLinkVerification] Zero CrossLinks in the header") - return nil - } - crossLinks := &types.CrossLinks{} - err := rlp.DecodeBytes(block.Header().CrossLinks(), crossLinks) - if err != nil { - return ctxerror.New("[CrossLinkVerification] failed to decode cross links", - "blockHash", block.Hash(), - "crossLinks", len(block.Header().CrossLinks()), - ).WithCause(err) - } - - if !crossLinks.IsSorted() { - return ctxerror.New("[CrossLinkVerification] cross links are not sorted", - "blockHash", block.Hash(), - "crossLinks", len(block.Header().CrossLinks()), - ) - } - - for _, crossLink := range *crossLinks { - cl, err := node.Blockchain().ReadCrossLink(crossLink.ShardID(), crossLink.BlockNum()) - if err == nil && cl != nil { - // Add slash for exist same blocknum but different crosslink - return ctxerror.New("crosslink already exist!") - } - if err = node.VerifyCrossLink(crossLink); err != nil { - return ctxerror.New("cannot VerifyBlockCrossLinks", - "blockHash", block.Hash(), - "blockNum", block.Number(), - "crossLinkShard", crossLink.ShardID(), - "crossLinkBlock", crossLink.BlockNum(), - "numTx", len(block.Transactions()), - ).WithCause(err) - } - } - return nil -} - -// ProcessCrossLinkMessage verify and process Node/CrossLink message into crosslink when it's valid -func (node *Node) ProcessCrossLinkMessage(msgPayload []byte) { - if node.NodeConfig.ShardID == 0 { - var crosslinks []types.CrossLink - err := rlp.DecodeBytes(msgPayload, &crosslinks) - if err != nil { - utils.Logger().Error(). - Err(err). - Msg("[ProcessingCrossLink] Crosslink Message Broadcast Unable to Decode") - return - } - - candidates := []types.CrossLink{} - utils.Logger().Debug(). - Msgf("[ProcessingCrossLink] Received crosslinks: %d", len(crosslinks)) - - for _, cl := range crosslinks { - exist, err := node.Blockchain().ReadCrossLink(cl.ShardID(), cl.Number().Uint64()) - if err == nil && exist != nil { - // TODO: leader add double sign checking - utils.Logger().Err(err). - Msgf("[ProcessingCrossLink] Cross Link already exists, pass. Block num: %d, shardID %d", cl.Number(), cl.ShardID()) - continue - } - - if err = node.VerifyCrossLink(cl); err != nil { - utils.Logger().Err(err). - Msgf("[ProcessingCrossLink] Failed to verify new cross link for blockNum %d epochNum %d shard %d skipped: %v", cl.BlockNum(), cl.Epoch().Uint64(), cl.ShardID(), cl) - continue - } - - candidates = append(candidates, cl) - utils.Logger().Debug(). - Msgf("[ProcessingCrossLink] Committing for shardID %d, blockNum %d", cl.ShardID(), cl.Number().Uint64()) - } - node.pendingCLMutex.Lock() - Len, _ := node.Blockchain().AddPendingCrossLinks(candidates) - node.pendingCLMutex.Unlock() - utils.Logger().Debug(). - Msgf("[ProcessingCrossLink] Add pending crosslinks, total pending: %d", Len) - } -} - func (node *Node) verifyIncomingReceipts(block *types.Block) error { m := make(map[common.Hash]struct{}) cxps := block.IncomingReceipts() @@ -218,69 +130,6 @@ func (node *Node) verifyIncomingReceipts(block *types.Block) error { return nil } -// VerifyCrossLink verifies the header is valid -func (node *Node) VerifyCrossLink(cl types.CrossLink) error { - if node.Blockchain().ShardID() != shard.BeaconChainShardID { - return ctxerror.New("Shard chains should not verify cross links") - } - - if cl.BlockNum() <= 1 { - return ctxerror.New("CrossLink BlockNumber should greater than 1") - } - - if !node.Blockchain().Config().IsCrossLink(cl.Epoch()) { - return ctxerror.New("CrossLink Epoch should >= cross link starting epoch", "crossLinkEpoch", cl.Epoch(), "cross_link_starting_eoch", node.Blockchain().Config().CrossLinkEpoch) - } - - // Verify signature of the new cross link header - // TODO: check whether to recalculate shard state - shardState, err := node.Blockchain().ReadShardState(cl.Epoch()) - committee := shardState.FindCommitteeByID(cl.ShardID()) - - if err != nil || committee == nil { - return ctxerror.New("[CrossLink] Failed to read shard state for cross link", "shardID", cl.ShardID(), "blockNum", cl.BlockNum()).WithCause(err) - } - var committerKeys []*bls.PublicKey - - parseKeysSuccess := true - for _, member := range committee.Slots { - committerKey := new(bls.PublicKey) - err = member.BlsPublicKey.ToLibBLSPublicKey(committerKey) - if err != nil { - parseKeysSuccess = false - break - } - committerKeys = append(committerKeys, committerKey) - } - if !parseKeysSuccess { - return ctxerror.New("[CrossLink] cannot convert BLS public key", "shardID", cl.ShardID(), "blockNum", cl.BlockNum()).WithCause(err) - } - - mask, err := bls_cosi.NewMask(committerKeys, nil) - if err != nil { - return ctxerror.New("cannot create group sig mask", "shardID", cl.ShardID(), "blockNum", cl.BlockNum()).WithCause(err) - } - if err := mask.SetMask(cl.Bitmap()); err != nil { - return ctxerror.New("cannot set group sig mask bits", "shardID", cl.ShardID(), "blockNum", cl.BlockNum()).WithCause(err) - } - - aggSig := bls.Sign{} - sig := cl.Signature() - err = aggSig.Deserialize(sig[:]) - if err != nil { - return ctxerror.New("unable to deserialize multi-signature from payload").WithCause(err) - } - - hash := cl.Hash() - blockNumBytes := make([]byte, 8) - binary.LittleEndian.PutUint64(blockNumBytes, cl.BlockNum()) - commitPayload := append(blockNumBytes, hash[:]...) - if !aggSig.VerifyHash(mask.AggregatePublic, commitPayload) { - return ctxerror.New("Failed to verify the signature for cross link", "shardID", cl.ShardID(), "blockNum", cl.BlockNum()) - } - return nil -} - // ProcessReceiptMessage store the receipts and merkle proof in local data store func (node *Node) ProcessReceiptMessage(msgPayload []byte) { cxp := types.CXReceiptsProof{} diff --git a/node/node_handler.go b/node/node_handler.go index 97d1150d4..0bde8b478 100644 --- a/node/node_handler.go +++ b/node/node_handler.go @@ -25,10 +25,6 @@ import ( libp2p_peer "github.com/libp2p/go-libp2p-core/peer" ) -const ( - crossLinkBatchSize = 3 -) - // receiveGroupMessage use libp2p pubsub mechanism to receive broadcast messages func (node *Node) receiveGroupMessage( receiver p2p.GroupReceiver, rxQueue msgq.MessageAdder, diff --git a/node/node_newblock.go b/node/node_newblock.go index 3abd8eede..b5a79b3df 100644 --- a/node/node_newblock.go +++ b/node/node_newblock.go @@ -139,7 +139,7 @@ func (node *Node) proposeNewBlock() (*types.Block, error) { // Prepare cross links var crossLinksToPropose types.CrossLinks - if node.NodeConfig.ShardID == 0 && node.Blockchain().Config().IsStaking(node.Worker.GetCurrentHeader().Epoch()) { + if node.NodeConfig.ShardID == 0 && node.Blockchain().Config().IsCrossLink(node.Worker.GetCurrentHeader().Epoch()) { node.pendingCLMutex.Lock() allPending, err := node.Blockchain().ReadPendingCrossLinks() node.pendingCLMutex.Unlock()