Merge pull request #217 from tomochain/revert-216-revert-211-double-validation

Adding double validation layer

consensus/posv/posv.go

@@ -444,6 +444,17 @@ func (c *Posv) GetMasternodes(chain consensus.ChainReader, header *types.Header)
 
 func (c *Posv) GetPeriod() uint64 { return c.config.Period }
 
+func WhoIsCreator(snap *Snapshot, header *types.Header) (common.Address, error) {
+	if header.Number.Uint64() == 0 {
+		return common.Address{}, errors.New("Don't take block 0")
+	}
+	m, err := ecrecover(header, snap.sigcache)
+	if err != nil {
+		return common.Address{}, err
+	}
+	return m, nil
+}
+
 func YourTurn(masternodes []common.Address, snap *Snapshot, header *types.Header, cur common.Address) (int, int, bool, error) {
 	if len(masternodes) == 0 {
 		return -1, -1, true, nil
@@ -453,7 +464,7 @@ func YourTurn(masternodes []common.Address, snap *Snapshot, header *types.Header
 	var err error
 	preIndex := -1
 	if header.Number.Uint64() != 0 {
-		pre, err = ecrecover(header, snap.sigcache)
+		pre, err = WhoIsCreator(snap, header)
 		if err != nil {
 			return 0, 0, false, err
 		}

contracts/utils.go

@@ -272,6 +272,7 @@ func ExtractValidatorsFromBytes(byteValidators []byte) []int64 {
 		intNumber, err := strconv.Atoi(string(trimByte))
 		if err != nil {
 			log.Error("Can not convert string to integer", "error", err)
+			return []int64{}
 		}
 		validators = append(validators, int64(intNumber))
 	}
@@ -568,26 +569,22 @@ func GetMasternodesFromCheckpointHeader(checkpointHeader *types.Header) []common
 }
 
 // Get m2 list from checkpoint block.
-func GetM2FromCheckpointBlock(checkpointBlock types.Block) ([]common.Address, error) {
+func GetM1M2FromCheckpointBlock(checkpointBlock *types.Block) (map[common.Address]common.Address, error) {
 	if checkpointBlock.Number().Int64()%common.EpocBlockRandomize != 0 {
 		return nil, errors.New("This block is not checkpoint block epoc.")
 	}
-
+	m1m2 := map[common.Address]common.Address{}
-	// Get singers from this block.
+	// Get signers from this block.
 	masternodes := GetMasternodesFromCheckpointHeader(checkpointBlock.Header())
 	validators := ExtractValidatorsFromBytes(checkpointBlock.Header().Validators)
 
-	var m2List []common.Address
+	if len(validators) < len(masternodes) {
-	lenMasternodes := len(masternodes)
+		return nil, errors.New("len(m2) is less than len(m1)")
-	var valAddr common.Address
-	for validatorIndex := range validators {
-		if validatorIndex < lenMasternodes {
-			valAddr = masternodes[validatorIndex]
-		} else {
-			valAddr = masternodes[validatorIndex-lenMasternodes]
-		}
-		m2List = append(m2List, valAddr)
 	}
-
+	if len(masternodes) > 0 {
-	return m2List, nil
+		for i, m1 := range masternodes {
+			m1m2[m1] = masternodes[validators[i]%int64(len(masternodes))]
+		}
+	}
+	return m1m2, nil
 }

core/blockchain.go

@@ -1635,16 +1635,16 @@ func (bc *BlockChain) UpdateM1() error {
 			ms = append(ms, posv.Masternode{Address: candidate, Stake: v.Uint64()})
 		}
 	}
-	log.Info("Ordered list of masternode candidates")
-	for _, m := range ms {
-		log.Info("", "address", m.Address.String(), "stake", m.Stake)
-	}
 	if len(ms) == 0 {
 		log.Info("No masternode candidates found. Keep the current masternodes set for the next epoch")
 	} else {
 		sort.Slice(ms, func(i, j int) bool {
 			return ms[i].Stake >= ms[j].Stake
 		})
+		log.Info("Ordered list of masternode candidates")
+		for _, m := range ms {
+			log.Info("", "address", m.Address.String(), "stake", m.Stake)
+		}
 		// update masternodes
 		log.Info("Updating new set of masternodes")
 		if len(ms) > common.MaxMasternodes {

core/tx_pool.go

@@ -562,6 +562,14 @@ func (pool *TxPool) local() map[common.Address]types.Transactions {
 	return txs
 }
 
+func (pool *TxPool) GetSender(tx *types.Transaction) (common.Address, error) {
+	from, err := types.Sender(pool.signer, tx)
+	if err != nil {
+		return common.Address{}, ErrInvalidSender
+	}
+	return from, nil
+}
+
 // validateTx checks whether a transaction is valid according to the consensus
 // rules and adheres to some heuristic limits of the local node (price and size).
 func (pool *TxPool) validateTx(tx *types.Transaction, local bool) error {

eth/backend.go

@@ -24,6 +24,7 @@ import (
 	"runtime"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"bytes"
 	"github.com/ethereum/go-ethereum/accounts"
@@ -54,8 +55,6 @@ import (
 	"github.com/ethereum/go-ethereum/rpc"
 )
 
-const NumOfMasternodes = 99
-
 type LesServer interface {
 	Start(srvr *p2p.Server)
 	Stop()
@@ -190,25 +189,65 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
 	if eth.chainConfig.Posv != nil {
 		c := eth.engine.(*posv.Posv)
 
-		// Hook sends tx sign to smartcontract after inserting block to chain.
+		// Hook double validation
-		importedHook := func(block *types.Block) {
+		doubleValidateHook := func(block *types.Block) error {
 			snap, err := c.GetSnapshot(eth.blockchain, block.Header())
 			if err != nil {
 				if err == consensus.ErrUnknownAncestor {
 					log.Warn("Block chain forked.", "error", err)
-				} else {
-					log.Error("Fail to get snapshot for sign tx validator.", "error", err)
 				}
-				return
+				return fmt.Errorf("Fail to get snapshot for sign tx validator: %v", err)
 			}
 			if _, authorized := snap.Signers[eth.etherbase]; authorized {
-				if err := contracts.CreateTransactionSign(chainConfig, eth.txPool, eth.accountManager, block, chainDb); err != nil {
+				m2, err := getM2(snap, eth, block)
-					log.Error("Fail to create tx sign for imported block", "error", err)
+				if err != nil {
-					return
+					return fmt.Errorf("Fail to validate M2 condition for importing block: %v", err)
 				}
+				if eth.etherbase != m2 {
+					// firstly, look into pending txPool
+					pendingMap, err := eth.txPool.Pending()
+					if err != nil {
+						log.Warn("Fail to get txPool pending", "err", err, "Continue with empty txPool pending.")
+						//reset pendingMap
+						pendingMap = map[common.Address]types.Transactions{}
+					}
+					txsSentFromM2 := pendingMap[m2]
+					if len(txsSentFromM2) > 0 {
+						for _, tx := range txsSentFromM2 {
+							if tx.To().String() == common.BlockSigners {
+								return nil
+							}
+						}
+					}
+					//then wait until signTx from m2 comes into txPool
+					txCh := make(chan core.TxPreEvent, txChanSize)
+					subEvent := eth.txPool.SubscribeTxPreEvent(txCh)
+					select {
+					case event := <-txCh:
+						from, err := eth.txPool.GetSender(event.Tx)
+						if (err == nil) && (event.Tx.To().String() == common.BlockSigners) && (from == m2) {
+							return nil
+						}
+					//timeout 10s
+					case <-time.After(time.Duration(10) * time.Second):
+						return fmt.Errorf("Time out waiting for confirmation from m2")
+					}
+					subEvent.Unsubscribe()
+				}
+				return nil
 			}
+			return fmt.Errorf("This address is not authorized to validate block")
 		}
-		eth.protocolManager.fetcher.SetImportedHook(importedHook)
+
+		signHook := func(block *types.Block) error {
+			if err := contracts.CreateTransactionSign(chainConfig, eth.txPool, eth.accountManager, block, chainDb); err != nil {
+				return fmt.Errorf("Fail to create tx sign for importing block: %v", err)
+			}
+			return nil
+		}
+
+		eth.protocolManager.fetcher.SetDoubleValidateHook(doubleValidateHook)
+		eth.protocolManager.fetcher.SetSignHook(signHook)
 
 		// Hook prepares validators M2 for the current epoch
 		c.HookValidator = func(header *types.Header, signers []common.Address) error {
@@ -329,6 +368,28 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
 	return eth, nil
 }
 
+func getM2(snap *posv.Snapshot, eth *Ethereum, block *types.Block) (common.Address, error) {
+	epoch := eth.chainConfig.Posv.Epoch
+	no := block.NumberU64()
+	cpNo := no
+	if no%epoch != 0 {
+		cpNo = no - (no % epoch)
+	}
+	if cpNo == 0 {
+		return eth.etherbase, nil
+	}
+	cpBlk := eth.blockchain.GetBlockByNumber(cpNo)
+	m, err := contracts.GetM1M2FromCheckpointBlock(cpBlk)
+	if err != nil {
+		return common.Address{}, err
+	}
+	m1, err := posv.WhoIsCreator(snap, block.Header())
+	if err != nil {
+		return common.Address{}, err
+	}
+	return m[m1], nil
+}
+
 func makeExtraData(extra []byte) []byte {
 	if len(extra) == 0 {
 		// create default extradata

eth/fetcher/fetcher.go

@@ -142,7 +142,8 @@ type Fetcher struct {
 	queueChangeHook    func(common.Hash, bool) // Method to call upon adding or deleting a block from the import queue
 	fetchingHook       func([]common.Hash)     // Method to call upon starting a block (eth/61) or header (eth/62) fetch
 	completingHook     func([]common.Hash)     // Method to call upon starting a block body fetch (eth/62)
-	importedHook       func(*types.Block)      // Method to call upon successful block import (both eth/61 and eth/62)
+	doubleValidateHook func(*types.Block) error
+	signHook           func(*types.Block) error
 }
 
 // New creates a block fetcher to retrieve blocks based on hash announcements.
@@ -665,19 +666,31 @@ func (f *Fetcher) insert(peer string, block *types.Block) {
 			f.dropPeer(peer)
 			return
 		}
+		// Invoke the dv hook to run double validation layer
+		if f.doubleValidateHook != nil {
+			if err := f.doubleValidateHook(block); err != nil {
+				log.Error("Double validation failed", "err", err, "Discard this block!")
+				return
+			}
+		}
+
 		// Run the actual import and log any issues
 		if _, err := f.insertChain(types.Blocks{block}); err != nil {
 			log.Debug("Propagated block import failed", "peer", peer, "number", block.Number(), "hash", hash, "err", err)
 			return
 		}
+
+		if f.signHook != nil {
+			if err := f.signHook(block); err != nil {
+				log.Error("Can't sign the imported block", "err", err)
+				return
+			}
+		}
+
 		// If import succeeded, broadcast the block
 		propAnnounceOutTimer.UpdateSince(block.ReceivedAt)
 		go f.broadcastBlock(block, false)
 
-		// Invoke the testing hook if needed
-		if f.importedHook != nil {
-			f.importedHook(block)
-		}
 	}()
 }
 
@@ -735,7 +748,12 @@ func (f *Fetcher) forgetBlock(hash common.Hash) {
 	}
 }
 
-// Bind import hook when block imported into chain.
+// Bind double validate hook before block imported into chain.
-func (f *Fetcher) SetImportedHook(importedHook func(*types.Block)) {
+func (f *Fetcher) SetDoubleValidateHook(doubleValidateHook func(*types.Block) error) {
-	f.importedHook = importedHook
+	f.doubleValidateHook = doubleValidateHook
+}
+
+// Bind double validate hook before block imported into chain.
+func (f *Fetcher) SetSignHook(signHook func(*types.Block) error) {
+	f.signHook = signHook
 }

eth/fetcher/fetcher_test.go

@@ -288,7 +288,10 @@ func testSequentialAnnouncements(t *testing.T, protocol int) {
 
 	// Iteratively announce blocks until all are imported
 	imported := make(chan *types.Block)
-	tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
+	tester.fetcher.signHook = func(block *types.Block) error {
+		imported <- block
+		return nil
+	}
 
 	for i := len(hashes) - 2; i >= 0; i-- {
 		tester.fetcher.Notify("valid", hashes[i], uint64(len(hashes)-i-1), time.Now().Add(-arriveTimeout), headerFetcher, bodyFetcher)
@@ -326,7 +329,10 @@ func testConcurrentAnnouncements(t *testing.T, protocol int) {
 	}
 	// Iteratively announce blocks until all are imported
 	imported := make(chan *types.Block)
-	tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
+	tester.fetcher.signHook = func(block *types.Block) error {
+		imported <- block
+		return nil
+	}
 
 	for i := len(hashes) - 2; i >= 0; i-- {
 		tester.fetcher.Notify("first", hashes[i], uint64(len(hashes)-i-1), time.Now().Add(-arriveTimeout), firstHeaderWrapper, firstBodyFetcher)
@@ -363,7 +369,10 @@ func testOverlappingAnnouncements(t *testing.T, protocol int) {
 	for i := 0; i < overlap; i++ {
 		imported <- nil
 	}
-	tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
+	tester.fetcher.signHook = func(block *types.Block) error {
+		imported <- block
+		return nil
+	}
 
 	for i := len(hashes) - 2; i >= 0; i-- {
 		tester.fetcher.Notify("valid", hashes[i], uint64(len(hashes)-i-1), time.Now().Add(-arriveTimeout), headerFetcher, bodyFetcher)
@@ -437,7 +446,10 @@ func testRandomArrivalImport(t *testing.T, protocol int) {
 
 	// Iteratively announce blocks, skipping one entry
 	imported := make(chan *types.Block, len(hashes)-1)
-	tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
+	tester.fetcher.signHook = func(block *types.Block) error {
+		imported <- block
+		return nil
+	}
 
 	for i := len(hashes) - 1; i >= 0; i-- {
 		if i != skip {
@@ -468,7 +480,10 @@ func testQueueGapFill(t *testing.T, protocol int) {
 
 	// Iteratively announce blocks, skipping one entry
 	imported := make(chan *types.Block, len(hashes)-1)
-	tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
+	tester.fetcher.signHook = func(block *types.Block) error {
+		imported <- block
+		return nil
+	}
 
 	for i := len(hashes) - 1; i >= 0; i-- {
 		if i != skip {
@@ -505,7 +520,10 @@ func testImportDeduplication(t *testing.T, protocol int) {
 	fetching := make(chan []common.Hash)
 	imported := make(chan *types.Block, len(hashes)-1)
 	tester.fetcher.fetchingHook = func(hashes []common.Hash) { fetching <- hashes }
-	tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
+	tester.fetcher.signHook = func(block *types.Block) error {
+		imported <- block
+		return nil
+	}
 
 	// Announce the duplicating block, wait for retrieval, and also propagate directly
 	tester.fetcher.Notify("valid", hashes[0], 1, time.Now().Add(-arriveTimeout), headerFetcher, bodyFetcher)
@@ -614,7 +632,10 @@ func testInvalidNumberAnnouncement(t *testing.T, protocol int) {
 	badBodyFetcher := tester.makeBodyFetcher("bad", blocks, 0)
 
 	imported := make(chan *types.Block)
-	tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
+	tester.fetcher.signHook = func(block *types.Block) error {
+		imported <- block
+		return nil
+	}
 
 	// Announce a block with a bad number, check for immediate drop
 	tester.fetcher.Notify("bad", hashes[0], 2, time.Now().Add(-arriveTimeout), badHeaderFetcher, badBodyFetcher)
@@ -666,7 +687,10 @@ func testEmptyBlockShortCircuit(t *testing.T, protocol int) {
 	tester.fetcher.completingHook = func(hashes []common.Hash) { completing <- hashes }
 
 	imported := make(chan *types.Block)
-	tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
+	tester.fetcher.signHook = func(block *types.Block) error {
+		imported <- block
+		return nil
+	}
 
 	// Iteratively announce blocks until all are imported
 	for i := len(hashes) - 2; i >= 0; i-- {
@@ -696,7 +720,10 @@ func testHashMemoryExhaustionAttack(t *testing.T, protocol int) {
 	tester := newTester()
 
 	imported, announces := make(chan *types.Block), int32(0)
-	tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
+	tester.fetcher.signHook = func(block *types.Block) error {
+		imported <- block
+		return nil
+	}
 	tester.fetcher.announceChangeHook = func(hash common.Hash, added bool) {
 		if added {
 			atomic.AddInt32(&announces, 1)
@@ -743,7 +770,10 @@ func TestBlockMemoryExhaustionAttack(t *testing.T) {
 	tester := newTester()
 
 	imported, enqueued := make(chan *types.Block), int32(0)
-	tester.fetcher.importedHook = func(block *types.Block) { imported <- block }
+	tester.fetcher.signHook = func(block *types.Block) error {
+		imported <- block
+		return nil
+	}
 	tester.fetcher.queueChangeHook = func(hash common.Hash, added bool) {
 		if added {
 			atomic.AddInt32(&enqueued, 1)