fix several small bugs and security issues on cross shard transaction

api/service/syncing/syncing.go

@@ -620,6 +620,7 @@ func (ss *StateSync) generateNewState(bc *core.BlockChain, worker *worker.Worker
 }
 
 // ProcessStateSync processes state sync from the blocks received but not yet processed so far
+// TODO: return error
 func (ss *StateSync) ProcessStateSync(startHash []byte, size uint32, bc *core.BlockChain, worker *worker.Worker) {
 	// Gets consensus hashes.
 	if !ss.GetConsensusHashes(startHash, size) {
@@ -733,6 +734,7 @@ func (ss *StateSync) SyncLoop(bc *core.BlockChain, worker *worker.Worker, willJo
 	if !isBeacon {
 		ss.RegisterNodeInfo()
 	}
+	// remove SyncLoopFrequency
 	ticker := time.NewTicker(SyncLoopFrequency * time.Second)
 	for {
 		select {

cmd/client/wallet/main.go

@@ -92,8 +92,8 @@ var (
 	transferReceiverPtr   = transferCommand.String("to", "", "Specify the receiver account")
 	transferAmountPtr     = transferCommand.Float64("amount", 0, "Specify the amount to transfer")
 	transferGasPricePtr   = transferCommand.Uint64("gasPrice", 0, "Specify the gas price amount. Unit is Nano.")
-	transferShardIDPtr    = transferCommand.Int("shardID", 0, "Specify the shard ID for the transfer")
-	transferToShardIDPtr  = transferCommand.Int("toShardID", 0, "Specify the destination shard ID for the transfer")
+	transferShardIDPtr    = transferCommand.Int("shardID", -1, "Specify the shard ID for the transfer")
+	transferToShardIDPtr  = transferCommand.Int("toShardID", -1, "Specify the destination shard ID for the transfer")
 	transferInputDataPtr  = transferCommand.String("inputData", "", "Base64-encoded input data to embed in the transaction")
 	transferSenderPassPtr = transferCommand.String("pass", "", "Passphrase of the sender's private key")
 

core/blockchain.go

@@ -1130,6 +1130,7 @@ func (bc *BlockChain) InsertChain(chain types.Blocks) (int, error) {
 				Str("parentHash", header.ParentHash.Hex()).
 				Msg("added block to chain")
 
+				// TODO: move into WriteBlockWithState
 			if header.ShardStateHash != (common.Hash{}) {
 				epoch := new(big.Int).Add(header.Epoch, common.Big1)
 				err = bc.WriteShardStateBytes(epoch, header.ShardState)
@@ -1138,6 +1139,8 @@ func (bc *BlockChain) InsertChain(chain types.Blocks) (int, error) {
 					return n, err
 				}
 			}
+
+			// TODO: move into WriteBlockWithState
 			if len(header.CrossLinks) > 0 {
 				crossLinks := &types.CrossLinks{}
 				err = rlp.DecodeBytes(header.CrossLinks, crossLinks)
@@ -2154,7 +2157,7 @@ func (bc *BlockChain) NextCXReceiptsCheckpoint(currentNum uint64, shardID uint32
 	for num := lastCheckpoint; num <= currentNum+1; num++ {
 		by, _ := rawdb.ReadCXReceiptsProofSpent(bc.db, shardID, num)
 		if by == rawdb.NAByte {
-			// TODO: check if there is IncompingReceiptsHash in crosslink header
+			// TODO chao: check if there is IncompingReceiptsHash in crosslink header
 			// if the rootHash is non-empty, it means incomingReceipts are not delivered
 			// otherwise, it means there is no cross-shard transactions for this block
 			newCheckpoint = num

core/chain_makers.go

@@ -96,7 +96,6 @@ func (b *BlockGen) AddTxWithChain(bc *BlockChain, tx *types.Transaction) {
 		b.SetCoinbase(common.Address{})
 	}
 	b.statedb.Prepare(tx.Hash(), common.Hash{}, len(b.txs))
-	// TODO (chao): may need to add cxReceipt for BlockGen
 	receipt, _, _, err := ApplyTransaction(b.config, bc, &b.header.Coinbase, b.gasPool, b.statedb, b.header, tx, &b.header.GasUsed, vm.Config{})
 	if err != nil {
 		panic(err)
@@ -186,7 +185,6 @@ func GenerateChain(config *params.ChainConfig, parent *types.Block, engine conse
 		}
 		if b.engine != nil {
 			// Finalize and seal the block
-			// TODO (chao): add cxReceipt in the last input
 			block, err := b.engine.Finalize(chainreader, b.header, statedb, b.txs, b.receipts, nil, nil)
 			if err != nil {
 				panic(err)

core/evm.go

@@ -95,7 +95,7 @@ func Transfer(db vm.StateDB, sender, recipient common.Address, amount *big.Int,
 	if txType == types.SameShardTx || txType == types.SubtractionOnly {
 		db.SubBalance(sender, amount)
 	}
-	if txType == types.SameShardTx || txType == types.AdditionOnly {
+	if txType == types.SameShardTx {
 		db.AddBalance(recipient, amount)
 	}
 }

core/state_processor.go

@@ -85,7 +85,10 @@ func (p *StateProcessor) Process(block *types.Block, statedb *state.DB, cfg vm.C
 
 	// incomingReceipts should always be processed after transactions (to be consistent with the block proposal)
 	for _, cx := range block.IncomingReceipts() {
-		ApplyIncomingReceipt(p.config, statedb, header, cx)
+		err := ApplyIncomingReceipt(p.config, statedb, header, cx)
+		if err != nil {
+			return nil, nil, nil, 0, ctxerror.New("cannot apply incoming receipts").WithCause(err)
+		}
 	}
 
 	// Finalize the block, applying any consensus engine specific extras (e.g. block rewards)
@@ -102,7 +105,9 @@ func getTransactionType(header *types.Header, tx *types.Transaction) types.Trans
 	if tx.ShardID() == tx.ToShardID() && header.ShardID == tx.ShardID() {
 		return types.SameShardTx
 	}
-	if tx.ShardID() != tx.ToShardID() && header.ShardID == tx.ShardID() {
+	numShards := ShardingSchedule.InstanceForEpoch(header.Epoch).NumShards()
+	// Assuming here all the shards are consecutive from 0 to n-1, n is total number of shards
+	if tx.ShardID() != tx.ToShardID() && header.ShardID == tx.ShardID() && tx.ToShardID() < numShards {
 		return types.SubtractionOnly
 	}
 	return types.InvalidTx
@@ -167,16 +172,15 @@ func ApplyTransaction(config *params.ChainConfig, bc ChainContext, author *commo
 }
 
 // ApplyIncomingReceipt will add amount into ToAddress in the receipt
-func ApplyIncomingReceipt(config *params.ChainConfig, db *state.DB, header *types.Header, cxp *types.CXReceiptsProof) {
+func ApplyIncomingReceipt(config *params.ChainConfig, db *state.DB, header *types.Header, cxp *types.CXReceiptsProof) error {
 	if cxp == nil {
-		return
+		return nil
 	}
 
 	// TODO: how to charge gas here?
 	for _, cx := range cxp.Receipts {
 		if cx == nil || cx.To == nil { // should not happend
-			utils.Logger().Warn().Msg("ApplyIncomingReceipts: Invalid incoming receipt!!")
-			continue
+			return ctxerror.New("ApplyIncomingReceipts: Invalid incomingReceipt!", "receipt", cx)
 		}
 		utils.Logger().Info().Msgf("ApplyIncomingReceipts: ADDING BALANCE %d", cx.Amount)
 
@@ -186,4 +190,5 @@ func ApplyIncomingReceipt(config *params.ChainConfig, db *state.DB, header *type
 		db.AddBalance(*cx.To, cx.Amount)
 		db.IntermediateRoot(config.IsEIP158(header.Number)).Bytes()
 	}
+	return nil
 }

core/types/block.go

@@ -473,9 +473,10 @@ func (b *Block) WithBody(transactions []*Transaction, uncles []*Header, incoming
 		header:           CopyHeader(b.header),
 		transactions:     make([]*Transaction, len(transactions)),
 		uncles:           make([]*Header, len(uncles)),
-		incomingReceipts: incomingReceipts,
+		incomingReceipts: make([]*CXReceiptsProof, len(incomingReceipts)),
 	}
 	copy(block.transactions, transactions)
+	copy(block.incomingReceipts, incomingReceipts)
 	for i := range uncles {
 		block.uncles[i] = CopyHeader(uncles[i])
 	}

core/types/transaction.go

@@ -45,7 +45,6 @@ type TransactionType byte
 const (
 	SameShardTx     TransactionType = iota
 	SubtractionOnly                 // only subtract tokens from source shard account
-	AdditionOnly                    // only add tokens to destination shard account
 	InvalidTx
 )
 
@@ -64,8 +63,8 @@ func (txType TransactionType) String() string {
 		return "SameShardTx"
 	} else if txType == SubtractionOnly {
 		return "SubtractionOnly"
-	} else if txType == AdditionOnly {
-		return "AdditionOnly"
+	} else if txType == InvalidTx {
+		return "InvalidTx"
 	}
 	return "Unknown"
 }

core/types/transaction_signing.go

@@ -224,10 +224,10 @@ func (fs FrontierSigner) Sender(tx *Transaction) (common.Address, error) {
 }
 
 func recoverPlain(sighash common.Hash, R, S, Vb *big.Int, homestead bool) (common.Address, error) {
-	V := byte(Vb.Uint64() - 27)
 	if Vb.BitLen() > 8 {
 		return common.Address{}, ErrInvalidSig
 	}
+	V := byte(Vb.Uint64() - 27)
 	if !crypto.ValidateSignatureValues(V, R, S, homestead) {
 		return common.Address{}, ErrInvalidSig
 	}

node/node_cross_shard.go

@@ -115,6 +115,15 @@ func (node *Node) verifyIncomingReceipts(block *types.Block) error {
 			return err
 		}
 	}
+
+	incomingReceiptHash := types.EmptyRootHash
+	if len(cxps) > 0 {
+		incomingReceiptHash = types.DeriveSha(cxps)
+	}
+	if incomingReceiptHash != block.Header().IncomingReceiptHash {
+		return ctxerror.New("[verifyIncomingReceipts] Invalid IncomingReceiptHash in block header")
+	}
+
 	return nil
 }
 
@@ -236,6 +245,7 @@ func (node *Node) ProposeCrossLinkDataForBeaconchain() (types.CrossLinks, error)
 					utils.Logger().Error().
 						Err(err).
 						Msgf("[CrossLink] Haven't received the first cross link %d", link.BlockNum().Uint64())
+					break
 				} else {
 					err := node.VerifyCrosslinkHeader(lastLink.Header(), link.Header())
 					if err != nil {
@@ -277,11 +287,7 @@ func (node *Node) ProcessReceiptMessage(msgPayload []byte) {
 		return
 	}
 
-	// TODO: check message signature is from the nodes of source shard.
-
-	// TODO: remove in future if not useful
-	node.Blockchain().WriteCXReceipts(cxp.MerkleProof.ShardID, cxp.MerkleProof.BlockNum.Uint64(), cxp.MerkleProof.BlockHash, cxp.Receipts, true)
-
 	utils.Logger().Debug().Interface("cxp", cxp).Msg("[ProcessReceiptMessage] Add CXReceiptsProof to pending Receipts")
+	// TODO: integrate with txpool
 	node.AddPendingReceipts(&cxp)
 }

node/node_handler.go

@@ -355,6 +355,7 @@ func (node *Node) VerifyNewBlock(newBlock *types.Block) error {
 	}
 
 	// Verify cross links
+	// TODO: move into ValidateNewBlock
 	if node.NodeConfig.ShardID == 0 {
 		err := node.VerifyBlockCrossLinks(newBlock)
 		if err != nil {
@@ -363,6 +364,7 @@ func (node *Node) VerifyNewBlock(newBlock *types.Block) error {
 		}
 	}
 
+	// TODO: move into ValidateNewBlock
 	err = node.verifyIncomingReceipts(newBlock)
 	if err != nil {
 		return ctxerror.New("[VerifyNewBlock] Cannot ValidateNewBlock", "blockHash", newBlock.Hash(),
@@ -566,6 +568,7 @@ func (node *Node) PostConsensusProcessing(newBlock *types.Block) {
 	}
 
 	// Update last consensus time for metrics
+	// TODO: randomly selected a few validators to broadcast messages instead of only leader broadcast
 	node.lastConsensusTime = time.Now().Unix()
 	if node.Consensus.PubKey.IsEqual(node.Consensus.LeaderPubKey) {
 		if node.NodeConfig.ShardID == 0 {
@@ -580,6 +583,7 @@ func (node *Node) PostConsensusProcessing(newBlock *types.Block) {
 			Msg("BINGO !!! Reached Consensus")
 	}
 
+	// TODO chao: Write New checkpoint after clean
 	node.Blockchain().CleanCXReceiptsCheckpointsByBlock(newBlock)
 
 	if node.NodeConfig.GetNetworkType() != nodeconfig.Mainnet {

node/worker/worker.go

@@ -93,7 +93,7 @@ func (w *Worker) SelectTransactionsForNewBlock(newBlockNum uint64, txs types.Tra
 	unselected := types.Transactions{}
 	invalid := types.Transactions{}
 	for _, tx := range txs {
-		if tx.ShardID() != w.shardID && tx.ToShardID() != w.shardID {
+		if tx.ShardID() != w.shardID {
 			invalid = append(invalid, tx)
 			continue
 		}
@@ -186,7 +186,10 @@ func (w *Worker) CommitReceipts(receiptsList []*types.CXReceiptsProof) error {
 	}
 
 	for _, cx := range receiptsList {
-		core.ApplyIncomingReceipt(w.config, w.current.state, w.current.header, cx)
+		err := core.ApplyIncomingReceipt(w.config, w.current.state, w.current.header, cx)
+		if err != nil {
+			return ctxerror.New("cannot apply receiptsList").WithCause(err)
+		}
 	}
 
 	for _, cx := range receiptsList {