diff --git a/consensus/posv/posv.go b/consensus/posv/posv.go index 3f63137c36..06c9c94367 100644 --- a/consensus/posv/posv.go +++ b/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 } diff --git a/contracts/utils.go b/contracts/utils.go index 435f6d668d..6046baed85 100644 --- a/contracts/utils.go +++ b/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.") } - - // Get singers from this block. + m1m2 := map[common.Address]common.Address{} + // Get signers from this block. masternodes := GetMasternodesFromCheckpointHeader(checkpointBlock.Header()) validators := ExtractValidatorsFromBytes(checkpointBlock.Header().Validators) - var m2List []common.Address - lenMasternodes := len(masternodes) - 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(validators) < len(masternodes) { + return nil, errors.New("len(m2) is less than len(m1)") } - - return m2List, nil + if len(masternodes) > 0 { + for i, m1 := range masternodes { + m1m2[m1] = masternodes[validators[i]%int64(len(masternodes))] + } + } + return m1m2, nil } diff --git a/core/blockchain.go b/core/blockchain.go index efd6c0758e..cf90d52177 100644 --- a/core/blockchain.go +++ b/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 { diff --git a/core/tx_pool.go b/core/tx_pool.go index d82619e952..e20aff5500 100644 --- a/core/tx_pool.go +++ b/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 { diff --git a/eth/backend.go b/eth/backend.go index be7e7275df..8b17d927ec 100644 --- a/eth/backend.go +++ b/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. - importedHook := func(block *types.Block) { + // Hook double validation + 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 { - log.Error("Fail to create tx sign for imported block", "error", err) - return + m2, err := getM2(snap, eth, block) + if err != nil { + 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 diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go index 4f0c916f7b..e75ecdbfa5 100644 --- a/eth/fetcher/fetcher.go +++ b/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. -func (f *Fetcher) SetImportedHook(importedHook func(*types.Block)) { - f.importedHook = importedHook +// Bind double validate hook before block imported into chain. +func (f *Fetcher) SetDoubleValidateHook(doubleValidateHook func(*types.Block) error) { + f.doubleValidateHook = doubleValidateHook +} + +// Bind double validate hook before block imported into chain. +func (f *Fetcher) SetSignHook(signHook func(*types.Block) error) { + f.signHook = signHook } diff --git a/eth/fetcher/fetcher_test.go b/eth/fetcher/fetcher_test.go index 9d53b98b60..af9a5a6b44 100644 --- a/eth/fetcher/fetcher_test.go +++ b/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)