[audit] Various fixes and todos for auditing; add log for collect rew…

…ards (#2448)

* More logs and checks before processing staking txns

* fix import

* Refactor block proposal

* Various fixes and todos for auditing; add log for collect rewards

* Fix lint

* fix comment

consensus/leader.go

@@ -311,6 +311,7 @@ func (consensus *Consensus) onCommit(msg *msg_pb.Message) {
 		return
 	}
 
+	// TODO(audit): verify signature on hash+blockNum+viewID (add a hard fork)
 	blockNumHash := make([]byte, 8)
 	binary.LittleEndian.PutUint64(blockNumHash, recvMsg.BlockNum)
 	commitPayload := append(blockNumHash, recvMsg.BlockHash[:]...)

consensus/threshold.go

@@ -37,6 +37,7 @@ func (consensus *Consensus) didReachPrepareQuorum() error {
 	consensus.aggregatedPrepareSig = aggSig
 	consensus.FBFTLog.AddMessage(FBFTMsg)
 	// Leader add commit phase signature
+	// TODO(audit): sign signature on hash+blockNum+viewID (add a hard fork)
 	blockNumHash := [8]byte{}
 	binary.LittleEndian.PutUint64(blockNumHash[:], consensus.blockNum)
 	commitPayload := append(blockNumHash[:], consensus.blockHash[:]...)

consensus/validator.go

@@ -196,7 +196,7 @@ func (consensus *Consensus) onPrepared(msg *msg_pb.Message) {
 	groupID := []nodeconfig.GroupID{nodeconfig.NewGroupIDByShardID(nodeconfig.ShardID(consensus.ShardID))}
 	for i, key := range consensus.PubKey.PublicKey {
 		networkMessage, _ := consensus.construct(
-			// TODO: should only sign on block hash
+			// TODO(audit): sign signature on hash+blockNum+viewID (add a hard fork)
 			msg_pb.MessageType_COMMIT,
 			append(blockNumBytes, consensus.blockHash[:]...),
 			key, consensus.priKey.PrivateKey[i],
@@ -249,6 +249,7 @@ func (consensus *Consensus) onCommitted(msg *msg_pb.Message) {
 		return
 	}
 
+	// TODO(audit): verify signature on hash+blockNum+viewID (add a hard fork)
 	blockNumBytes := make([]byte, 8)
 	binary.LittleEndian.PutUint64(blockNumBytes, recvMsg.BlockNum)
 	commitPayload := append(blockNumBytes, recvMsg.BlockHash[:]...)

consensus/view_change.go

@@ -377,6 +377,7 @@ func (consensus *Consensus) onViewChange(msg *msg_pb.Message) {
 			consensus.aggregatedPrepareSig = aggSig
 			consensus.prepareBitmap = mask
 			// Leader sign and add commit message
+			// TODO(audit): verify signature on hash+blockNum+viewID (add a hard fork)
 			blockNumBytes := [8]byte{}
 			binary.LittleEndian.PutUint64(blockNumBytes[:], consensus.blockNum)
 			commitPayload := append(blockNumBytes[:], consensus.blockHash[:]...)

core/blockchain.go

@@ -265,6 +265,7 @@ func (bc *BlockChain) ValidateNewBlock(block *types.Block) error {
 		return err
 	}
 
+	// Verify all the hash roots (state, txns, receipts, cross-shard)
 	if err := bc.Validator().ValidateState(
 		block, state, receipts, cxReceipts, usedGas,
 	); err != nil {

core/staking_verifier.go

@@ -4,11 +4,14 @@ import (
 	"bytes"
 	"math/big"
 
+	"github.com/harmony-one/harmony/internal/utils"
+
+	"github.com/pkg/errors"
+
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/harmony-one/harmony/core/vm"
 	common2 "github.com/harmony-one/harmony/internal/common"
 	staking "github.com/harmony-one/harmony/staking/types"
-	"github.com/pkg/errors"
 )
 
 var (
@@ -98,10 +101,9 @@ func VerifyAndEditValidatorFromMsg(
 		return nil, errCommissionRateChangeTooHigh
 	}
 
-	// TODO: make sure we are reading from the correct snapshot
 	snapshotValidator, err := chainContext.ReadValidatorSnapshot(wrapper.Address)
 	if err != nil {
-		return nil, err
+		return nil, errors.WithMessage(err, "Validator snapshot not found.")
 	}
 	rateAtBeginningOfEpoch := snapshotValidator.Validator.Rate
 
@@ -172,9 +174,14 @@ func VerifyAndDelegateFromMsg(
 				); err != nil {
 					return nil, nil, err
 				}
-				// Return remaining balance to be deducted for delegation
 				if delegateBalance.Cmp(big.NewInt(0)) < 0 {
-					return nil, nil, errInsufficientBalanceForStake // shouldn't really happen
+					return nil, nil, errNegativeAmount // shouldn't really happen
+				}
+				// Return remaining balance to be deducted for delegation
+				if !CanTransfer(stateDB, msg.DelegatorAddress, delegateBalance) {
+					return nil, nil, errors.Wrapf(
+						errInsufficientBalanceForStake, "had %v, tried to stake %v",
+						stateDB.GetBalance(msg.DelegatorAddress), delegateBalance)
 				}
 				return wrapper, delegateBalance, nil
 			}
@@ -189,7 +196,9 @@ func VerifyAndDelegateFromMsg(
 	}
 	// If no redelegation, create new delegation
 	if !CanTransfer(stateDB, msg.DelegatorAddress, msg.Amount) {
-		return nil, nil, errInsufficientBalanceForStake
+		return nil, nil, errors.Wrapf(
+			errInsufficientBalanceForStake, "had %v, tried to stake %v",
+			stateDB.GetBalance(msg.DelegatorAddress), msg.Amount)
 	}
 	wrapper.Delegations = append(
 		wrapper.Delegations, staking.NewDelegation(
@@ -270,8 +279,15 @@ func VerifyAndCollectRewardsFromDelegation(
 			delegation := &wrapper.Delegations[delegation.Index]
 			if delegation.Reward.Cmp(common.Big0) > 0 {
 				totalRewards.Add(totalRewards, delegation.Reward)
+				delegation.Reward.SetUint64(0)
 			}
-			delegation.Reward.SetUint64(0)
+		} else {
+			utils.Logger().Warn().
+				Str("validator", delegation.ValidatorAddress.String()).
+				Uint64("delegation index", delegation.Index).
+				Int("delegations length", len(wrapper.Delegations)).
+				Msg("Delegation index out of bound")
+			return nil, nil, errors.New("Delegation index out of bound")
 		}
 		if err := wrapper.SanityCheck(
 			staking.DoNotEnforceMaxBLS,

core/state_processor.go

@@ -99,7 +99,7 @@ func (p *StateProcessor) Process(
 		allLogs = append(allLogs, receipt.Logs...)
 	}
 
-	// Iterate over staking transactions
+	// Iterate over and process the staking transactions
 	L := len(block.Transactions())
 	for i, tx := range block.StakingTransactions() {
 		statedb.Prepare(tx.Hash(), block.Hash(), i+L)
@@ -119,14 +119,14 @@ func (p *StateProcessor) Process(
 		err := ApplyIncomingReceipt(p.config, statedb, header, cx)
 		if err != nil {
 			return nil, nil,
-				nil, 0, nil, ctxerror.New("cannot apply incoming receipts").WithCause(err)
+				nil, 0, nil, ctxerror.New("[Process] Cannot apply incoming receipts").WithCause(err)
 		}
 	}
 
 	slashes := slash.Records{}
 	if s := header.Slashes(); len(s) > 0 {
 		if err := rlp.DecodeBytes(s, &slashes); err != nil {
-			return nil, nil, nil, 0, nil, ctxerror.New("cannot finalize block").WithCause(err)
+			return nil, nil, nil, 0, nil, ctxerror.New("[Process] Cannot finalize block").WithCause(err)
 		}
 	}
 
@@ -136,7 +136,7 @@ func (p *StateProcessor) Process(
 		receipts, outcxs, incxs, block.StakingTransactions(), slashes,
 	)
 	if err != nil {
-		return nil, nil, nil, 0, nil, ctxerror.New("cannot finalize block").WithCause(err)
+		return nil, nil, nil, 0, nil, ctxerror.New("[Process] Cannot finalize block").WithCause(err)
 	}
 
 	return receipts, outcxs, allLogs, *usedGas, payout, nil
@@ -253,11 +253,8 @@ func ApplyStakingTransaction(
 
 	// Apply the transaction to the current state (included in the env)
 	gas, err = ApplyStakingMessage(vmenv, msg, gp, bc)
-	utils.Logger().Info().Msgf("ApplyStakingMessage: usedGas: %v, err: %v, stakingTxn:", gas, err)
-
-	// even there is error, we charge it
 	if err != nil {
-		return nil, gas, err
+		return nil, 0, err
 	}
 
 	// Update the state with pending changes
@@ -272,6 +269,11 @@ func ApplyStakingTransaction(
 	receipt.TxHash = tx.Hash()
 	receipt.GasUsed = gas
 
+	// TODO(audit): add more log to staking txns; expose them in block explorer.
+	if config.IsReceiptLog(header.Epoch()) {
+		receipt.Logs = statedb.GetLogs(tx.Hash())
+	}
+
 	return receipt, gas, nil
 }
 

core/state_transition.go

@@ -20,6 +20,9 @@ import (
 	"math"
 	"math/big"
 
+	staking2 "github.com/harmony-one/harmony/staking"
+	"github.com/harmony-one/harmony/staking/network"
+
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/rlp"
 	"github.com/harmony-one/harmony/core/types"
@@ -291,7 +294,6 @@ func (st *StateTransition) StakingTransitionDb() (usedGas uint64, err error) {
 	homestead := st.evm.ChainConfig().IsS3(st.evm.EpochNumber) // s3 includes homestead
 
 	// Pay intrinsic gas
-	// TODO: propose staking-specific formula for staking transaction
 	gas, err := IntrinsicGas(st.data, false, homestead, msg.Type() == types.StakeCreateVal)
 
 	if err != nil {
@@ -354,7 +356,15 @@ func (st *StateTransition) StakingTransitionDb() (usedGas uint64, err error) {
 		if msg.From() != stkMsg.DelegatorAddress {
 			return 0, errInvalidSigner
 		}
-		err = st.verifyAndApplyCollectRewards(stkMsg)
+		collectedRewards, err := st.verifyAndApplyCollectRewards(stkMsg)
+		if err == nil {
+			st.state.AddLog(&types.Log{
+				Address:     stkMsg.DelegatorAddress,
+				Topics:      []common.Hash{staking2.CollectRewardsTopic},
+				Data:        collectedRewards.Bytes(),
+				BlockNumber: st.evm.BlockNumber.Uint64(),
+			})
+		}
 	default:
 		return 0, staking.ErrInvalidStakingKind
 	}
@@ -373,10 +383,10 @@ func (st *StateTransition) verifyAndApplyCreateValidatorTx(
 	if err != nil {
 		return err
 	}
-	if err := st.state.UpdateValidatorWrapper(wrapper.Validator.Address, wrapper); err != nil {
+	if err := st.state.UpdateValidatorWrapper(wrapper.Address, wrapper); err != nil {
 		return err
 	}
-	st.state.SetValidatorFlag(wrapper.Validator.Address)
+	st.state.SetValidatorFlag(wrapper.Address)
 	st.state.SubBalance(wrapper.Address, createValidator.Amount)
 	return nil
 }
@@ -396,40 +406,40 @@ func (st *StateTransition) verifyAndApplyDelegateTx(delegate *staking.Delegate)
 	if err != nil {
 		return err
 	}
-	if err := st.state.UpdateValidatorWrapper(wrapper.Validator.Address, wrapper); err != nil {
-		return err
-	}
+
 	st.state.SubBalance(delegate.DelegatorAddress, balanceToBeDeducted)
-	return nil
+
+	return st.state.UpdateValidatorWrapper(wrapper.Address, wrapper)
 }
 
 func (st *StateTransition) verifyAndApplyUndelegateTx(undelegate *staking.Undelegate) error {
 	wrapper, err := VerifyAndUndelegateFromMsg(st.state, st.evm.EpochNumber, undelegate)
 	if err != nil {
 		return err
 	}
-	return st.state.UpdateValidatorWrapper(wrapper.Validator.Address, wrapper)
+	return st.state.UpdateValidatorWrapper(wrapper.Address, wrapper)
 }
 
-func (st *StateTransition) verifyAndApplyCollectRewards(collectRewards *staking.CollectRewards) error {
+func (st *StateTransition) verifyAndApplyCollectRewards(collectRewards *staking.CollectRewards) (*big.Int, error) {
 	if st.bc == nil {
-		return errors.New("[CollectRewards] No chain context provided")
+		return network.NoReward, errors.New("[CollectRewards] No chain context provided")
 	}
+	// TODO(audit): make sure the delegation index is always consistent with onchain data
 	delegations, err := st.bc.ReadDelegationsByDelegator(collectRewards.DelegatorAddress)
 	if err != nil {
-		return err
+		return network.NoReward, err
 	}
 	updatedValidatorWrappers, totalRewards, err := VerifyAndCollectRewardsFromDelegation(
 		st.state, delegations,
 	)
 	if err != nil {
-		return err
+		return network.NoReward, err
 	}
 	for _, wrapper := range updatedValidatorWrappers {
-		if err := st.state.UpdateValidatorWrapper(wrapper.Validator.Address, wrapper); err != nil {
-			return err
+		if err := st.state.UpdateValidatorWrapper(wrapper.Address, wrapper); err != nil {
+			return network.NoReward, err
 		}
 	}
 	st.state.AddBalance(collectRewards.DelegatorAddress, totalRewards)
-	return nil
+	return totalRewards, nil
 }

core/types/block.go

@@ -294,12 +294,19 @@ func NewBlock(
 	header *block.Header, txs []*Transaction,
 	receipts []*Receipt, outcxs []*CXReceipt, incxs []*CXReceiptsProof,
 	stks []*staking.StakingTransaction) *Block {
+
 	b := &Block{header: CopyHeader(header)}
 
 	if len(receipts) != len(txs)+len(stks) {
+		utils.Logger().Error().
+			Int("receiptsLen", len(receipts)).
+			Int("txnsLen", len(txs)).
+			Int("stakingTxnsLen", len(stks)).
+			Msg("Length of receipts doesn't match length of transactions")
 		return nil
 	}
 
+	// Put transactions into block
 	if len(txs) == 0 && len(stks) == 0 {
 		b.header.SetTxHash(EmptyRootHash)
 	} else {
@@ -315,15 +322,16 @@ func NewBlock(
 		))
 	}
 
+	// Put receipts into block
 	if len(receipts) == 0 {
 		b.header.SetReceiptHash(EmptyRootHash)
 	} else {
 		b.header.SetReceiptHash(DeriveSha(Receipts(receipts)))
 		b.header.SetBloom(CreateBloom(receipts))
 	}
 
+	// Put cross-shard receipts (ingres/egress) into block
 	b.header.SetOutgoingReceiptHash(CXReceipts(outcxs).ComputeMerkleRoot())
-
 	if len(incxs) == 0 {
 		b.header.SetIncomingReceiptHash(EmptyRootHash)
 	} else {
@@ -332,6 +340,7 @@ func NewBlock(
 		copy(b.incomingReceipts, incxs)
 	}
 
+	// Great! Block is finally finalized.
 	return b
 }
 

internal/chain/engine.go

@@ -208,6 +208,8 @@ func (e *engineImpl) VerifySeal(chain engine.ChainReader, header *block.Header)
 		if err != nil {
 			return errors.Wrapf(err, "cannot decoded shard state")
 		}
+
+		// TODO(audit): reuse a singleton decider and not recreate it for every single block
 		d := quorum.NewDecider(quorum.SuperMajorityStake)
 		d.SetMyPublicKeyProvider(func() (*multibls.PublicKey, error) {
 			return nil, nil
@@ -236,6 +238,7 @@ func (e *engineImpl) VerifySeal(chain engine.ChainReader, header *block.Header)
 		}
 	}
 
+	// TODO(audit): verify signature on hash+blockNum+viewID (add a hard fork)
 	blockNumHash := make([]byte, 8)
 	binary.LittleEndian.PutUint64(blockNumHash, header.Number().Uint64()-1)
 	lastCommitPayload := append(blockNumHash, parentHash[:]...)
@@ -289,6 +292,7 @@ func (e *engineImpl) Finalize(
 		}
 	}
 
+	// Finalize the state root
 	header.SetRoot(state.IntermediateRoot(chain.Config().IsS3(header.Epoch())))
 	return types.NewBlock(header, txs, receipts, outcxs, incxs, stks), payout, nil
 }
@@ -445,6 +449,8 @@ func (e *engineImpl) VerifyHeaderWithSignature(chain engine.ChainReader, header
 		if err != nil {
 			return errors.Wrapf(err, "cannot read shard state")
 		}
+
+		// TODO(audit): reuse a singleton decider and not recreate it for every single block
 		d := quorum.NewDecider(quorum.SuperMajorityStake)
 		d.SetMyPublicKeyProvider(func() (*multibls.PublicKey, error) {
 			return nil, nil
@@ -470,6 +476,7 @@ func (e *engineImpl) VerifyHeaderWithSignature(chain engine.ChainReader, header
 				"need", quorumCount, "got", count)
 		}
 	}
+	// TODO(audit): verify signature on hash+blockNum+viewID (add a hard fork)
 	blockNumHash := make([]byte, 8)
 	binary.LittleEndian.PutUint64(blockNumHash, header.Number().Uint64())
 	commitPayload := append(blockNumHash, hash[:]...)

node/node_explorer.go

@@ -52,6 +52,7 @@ func (node *Node) ExplorerMessageHandler(payload []byte) {
 			return
 		}
 
+		// TODO(audit): verify signature on hash+blockNum+viewID (add a hard fork)
 		blockNumHash := make([]byte, 8)
 		binary.LittleEndian.PutUint64(blockNumHash, recvMsg.BlockNum)
 		commitPayload := append(blockNumHash, recvMsg.BlockHash[:]...)