From 959dbec5a6f5d237212eaa27609f5393755011b1 Mon Sep 17 00:00:00 2001 From: Tuna Date: Fri, 12 Oct 2018 10:44:42 +0700 Subject: [PATCH 1/5] Revert "Revert "adding double validation layer"" --- consensus/posv/posv.go | 13 +++++++- contracts/utils.go | 27 ++++++++--------- core/blockchain.go | 8 ++--- core/tx_pool.go | 8 +++++ eth/backend.go | 68 +++++++++++++++++++++++++++++++++++++++++- eth/fetcher/fetcher.go | 2 +- 6 files changed, 104 insertions(+), 22 deletions(-) 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..4d2da192bf 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -52,6 +52,7 @@ import ( "github.com/ethereum/go-ethereum/params" "github.com/ethereum/go-ethereum/rlp" "github.com/ethereum/go-ethereum/rpc" + "time" ) const NumOfMasternodes = 99 @@ -202,10 +203,56 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { return } if _, authorized := snap.Signers[eth.etherbase]; authorized { - if err := contracts.CreateTransactionSign(chainConfig, eth.txPool, eth.accountManager, block, chainDb); err != nil { + // double validation + m2, err := getM2(snap, eth, block) + if err != nil { + log.Error("Fail to validate M2 condition for imported block", "error", err) + return + } + if eth.etherbase != m2 { + // firstly, look into pending txPool + pendingMap, err := eth.txPool.Pending() + if err != nil { + log.Error("Fail to get txPool pending", "err", err) + //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 { + 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 + } + return + } + } + } + //then wait until signTx from m2 comes into txPool + txCh := make(chan core.TxPreEvent, txChanSize) + subEvent := eth.txPool.SubscribeTxPreEvent(txCh) + G: + select { + case event := <-txCh: + from, err := eth.txPool.GetSender(event.Tx) + if (err == nil) && (event.Tx.To().String() == common.BlockSigners) && (from == m2) { + 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 + } + return + } + //timeout 10s + case <-time.After(time.Duration(10) * time.Second): + break G + } + subEvent.Unsubscribe() + } else 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 } + // end of double validation } } eth.protocolManager.fetcher.SetImportedHook(importedHook) @@ -329,6 +376,25 @@ 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) + } + 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..fefaf91c03 100644 --- a/eth/fetcher/fetcher.go +++ b/eth/fetcher/fetcher.go @@ -674,7 +674,7 @@ func (f *Fetcher) insert(peer string, block *types.Block) { propAnnounceOutTimer.UpdateSince(block.ReceivedAt) go f.broadcastBlock(block, false) - // Invoke the testing hook if needed + // Invoke the imported hook if needed if f.importedHook != nil { f.importedHook(block) } From 33a578824bffcb41db03867a33dfd3295c3d18a0 Mon Sep 17 00:00:00 2001 From: Tuna Date: Fri, 12 Oct 2018 11:52:59 +0700 Subject: [PATCH 2/5] correct order - dv before importing --- eth/backend.go | 32 +++++++++++------------- eth/fetcher/fetcher.go | 24 ++++++++++-------- eth/fetcher/fetcher_test.go | 50 +++++++++++++++++++++++++++++-------- 3 files changed, 68 insertions(+), 38 deletions(-) diff --git a/eth/backend.go b/eth/backend.go index 4d2da192bf..5f7fc7066b 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" @@ -52,11 +53,8 @@ import ( "github.com/ethereum/go-ethereum/params" "github.com/ethereum/go-ethereum/rlp" "github.com/ethereum/go-ethereum/rpc" - "time" ) -const NumOfMasternodes = 99 - type LesServer interface { Start(srvr *p2p.Server) Stop() @@ -192,28 +190,25 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { c := eth.engine.(*posv.Posv) // Hook sends tx sign to smartcontract after inserting block to chain. - importedHook := func(block *types.Block) { + importedHook := 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 { // double validation m2, err := getM2(snap, eth, block) if err != nil { - log.Error("Fail to validate M2 condition for imported block", "error", err) - 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.Error("Fail to get txPool pending", "err", err) + log.Warn("Fail to get txPool pending", "err", err, "Continue with empty txPool pending.") //reset pendingMap pendingMap = map[common.Address]types.Transactions{} } @@ -222,10 +217,9 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { for _, tx := range txsSentFromM2 { if tx.To().String() == common.BlockSigners { 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 + return fmt.Errorf("Fail to create tx sign for importing block: %v", err) } - return + return nil } } } @@ -238,10 +232,9 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { from, err := eth.txPool.GetSender(event.Tx) if (err == nil) && (event.Tx.To().String() == common.BlockSigners) && (from == m2) { 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 + return fmt.Errorf("Fail to create tx sign for importing block: %v", err) } - return + return nil } //timeout 10s case <-time.After(time.Duration(10) * time.Second): @@ -249,11 +242,11 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { } subEvent.Unsubscribe() } else 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 + return fmt.Errorf("Fail to create tx sign for importing block: %v", err) } // end of double validation } + return nil } eth.protocolManager.fetcher.SetImportedHook(importedHook) @@ -383,6 +376,9 @@ func getM2(snap *posv.Snapshot, eth *Ethereum, block *types.Block) (common.Addre 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 { diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go index fefaf91c03..94ac16e405 100644 --- a/eth/fetcher/fetcher.go +++ b/eth/fetcher/fetcher.go @@ -138,11 +138,11 @@ type Fetcher struct { dropPeer peerDropFn // Drops a peer for misbehaving // Testing hooks - announceChangeHook func(common.Hash, bool) // Method to call upon adding or deleting a hash from the announce list - 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) + announceChangeHook func(common.Hash, bool) // Method to call upon adding or deleting a hash from the announce list + 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) error // Method to call upon successful block import (both eth/61 and eth/62) } // New creates a block fetcher to retrieve blocks based on hash announcements. @@ -665,6 +665,14 @@ func (f *Fetcher) insert(peer string, block *types.Block) { f.dropPeer(peer) return } + // Invoke the imported hook to run double validation layer + if f.importedHook != nil { + if err := f.importedHook(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) @@ -674,10 +682,6 @@ func (f *Fetcher) insert(peer string, block *types.Block) { propAnnounceOutTimer.UpdateSince(block.ReceivedAt) go f.broadcastBlock(block, false) - // Invoke the imported hook if needed - if f.importedHook != nil { - f.importedHook(block) - } }() } @@ -736,6 +740,6 @@ func (f *Fetcher) forgetBlock(hash common.Hash) { } // Bind import hook when block imported into chain. -func (f *Fetcher) SetImportedHook(importedHook func(*types.Block)) { +func (f *Fetcher) SetImportedHook(importedHook func(*types.Block) error) { f.importedHook = importedHook } diff --git a/eth/fetcher/fetcher_test.go b/eth/fetcher/fetcher_test.go index 9d53b98b60..fc16a0f5f3 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.importedHook = 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.importedHook = 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.importedHook = 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.importedHook = 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.importedHook = 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.importedHook = 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.importedHook = 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.importedHook = 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.importedHook = 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.importedHook = func(block *types.Block) error { + imported <- block + return nil + } tester.fetcher.queueChangeHook = func(hash common.Hash, added bool) { if added { atomic.AddInt32(&enqueued, 1) From ab566a333a7ab1598ebbff65c996e13a70849a35 Mon Sep 17 00:00:00 2001 From: Tuna Date: Wed, 17 Oct 2018 16:21:40 +0700 Subject: [PATCH 3/5] double validate, then import, then create txSign --- eth/backend.go | 29 ++++++++++++++--------------- eth/fetcher/fetcher.go | 36 +++++++++++++++++++++++++----------- eth/fetcher/fetcher_test.go | 20 ++++++++++---------- 3 files changed, 49 insertions(+), 36 deletions(-) diff --git a/eth/backend.go b/eth/backend.go index 5f7fc7066b..8b17d927ec 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -189,8 +189,8 @@ 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) error { + // Hook double validation + doubleValidateHook := func(block *types.Block) error { snap, err := c.GetSnapshot(eth.blockchain, block.Header()) if err != nil { if err == consensus.ErrUnknownAncestor { @@ -199,7 +199,6 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { return fmt.Errorf("Fail to get snapshot for sign tx validator: %v", err) } if _, authorized := snap.Signers[eth.etherbase]; authorized { - // double validation m2, err := getM2(snap, eth, block) if err != nil { return fmt.Errorf("Fail to validate M2 condition for importing block: %v", err) @@ -216,9 +215,6 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { if len(txsSentFromM2) > 0 { for _, tx := range txsSentFromM2 { if tx.To().String() == common.BlockSigners { - 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 } } @@ -226,29 +222,32 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { //then wait until signTx from m2 comes into txPool txCh := make(chan core.TxPreEvent, txChanSize) subEvent := eth.txPool.SubscribeTxPreEvent(txCh) - G: select { case event := <-txCh: from, err := eth.txPool.GetSender(event.Tx) if (err == nil) && (event.Tx.To().String() == common.BlockSigners) && (from == m2) { - 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 } //timeout 10s case <-time.After(time.Duration(10) * time.Second): - break G + return fmt.Errorf("Time out waiting for confirmation from m2") } subEvent.Unsubscribe() - } else 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) } - // end of double validation + return nil + } + return fmt.Errorf("This address is not authorized to validate block") + } + + 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.SetImportedHook(importedHook) + + 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 { diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go index 94ac16e405..e75ecdbfa5 100644 --- a/eth/fetcher/fetcher.go +++ b/eth/fetcher/fetcher.go @@ -138,11 +138,12 @@ type Fetcher struct { dropPeer peerDropFn // Drops a peer for misbehaving // Testing hooks - announceChangeHook func(common.Hash, bool) // Method to call upon adding or deleting a hash from the announce list - 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) error // Method to call upon successful block import (both eth/61 and eth/62) + announceChangeHook func(common.Hash, bool) // Method to call upon adding or deleting a hash from the announce list + 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) + doubleValidateHook func(*types.Block) error + signHook func(*types.Block) error } // New creates a block fetcher to retrieve blocks based on hash announcements. @@ -665,9 +666,9 @@ func (f *Fetcher) insert(peer string, block *types.Block) { f.dropPeer(peer) return } - // Invoke the imported hook to run double validation layer - if f.importedHook != nil { - if err := f.importedHook(block); err != nil { + // 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 } @@ -678,6 +679,14 @@ func (f *Fetcher) insert(peer string, block *types.Block) { 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) @@ -739,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) error) { - 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 fc16a0f5f3..af9a5a6b44 100644 --- a/eth/fetcher/fetcher_test.go +++ b/eth/fetcher/fetcher_test.go @@ -288,7 +288,7 @@ 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) error { + tester.fetcher.signHook = func(block *types.Block) error { imported <- block return nil } @@ -329,7 +329,7 @@ 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) error { + tester.fetcher.signHook = func(block *types.Block) error { imported <- block return nil } @@ -369,7 +369,7 @@ func testOverlappingAnnouncements(t *testing.T, protocol int) { for i := 0; i < overlap; i++ { imported <- nil } - tester.fetcher.importedHook = func(block *types.Block) error { + tester.fetcher.signHook = func(block *types.Block) error { imported <- block return nil } @@ -446,7 +446,7 @@ 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) error { + tester.fetcher.signHook = func(block *types.Block) error { imported <- block return nil } @@ -480,7 +480,7 @@ 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) error { + tester.fetcher.signHook = func(block *types.Block) error { imported <- block return nil } @@ -520,7 +520,7 @@ 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) error { + tester.fetcher.signHook = func(block *types.Block) error { imported <- block return nil } @@ -632,7 +632,7 @@ 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) error { + tester.fetcher.signHook = func(block *types.Block) error { imported <- block return nil } @@ -687,7 +687,7 @@ 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) error { + tester.fetcher.signHook = func(block *types.Block) error { imported <- block return nil } @@ -720,7 +720,7 @@ func testHashMemoryExhaustionAttack(t *testing.T, protocol int) { tester := newTester() imported, announces := make(chan *types.Block), int32(0) - tester.fetcher.importedHook = func(block *types.Block) error { + tester.fetcher.signHook = func(block *types.Block) error { imported <- block return nil } @@ -770,7 +770,7 @@ func TestBlockMemoryExhaustionAttack(t *testing.T) { tester := newTester() imported, enqueued := make(chan *types.Block), int32(0) - tester.fetcher.importedHook = func(block *types.Block) error { + tester.fetcher.signHook = func(block *types.Block) error { imported <- block return nil } From 59f3d3a52e7f54a31e18b64f0784c2098201534c Mon Sep 17 00:00:00 2001 From: Tuna Date: Tue, 23 Oct 2018 16:06:26 +0700 Subject: [PATCH 4/5] get snapshot from parent blk --- eth/backend.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/eth/backend.go b/eth/backend.go index 8b17d927ec..4bea8f2d15 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -191,7 +191,8 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { // Hook double validation doubleValidateHook := func(block *types.Block) error { - snap, err := c.GetSnapshot(eth.blockchain, block.Header()) + parentBlk := eth.blockchain.GetBlockByHash(block.ParentHash()) + snap, err := c.GetSnapshot(eth.blockchain, parentBlk.Header()) if err != nil { if err == consensus.ErrUnknownAncestor { log.Warn("Block chain forked.", "error", err) From 1afaa5f89e705534d0d4bc8ed4bd5df730de9e2b Mon Sep 17 00:00:00 2001 From: Nguyen Ba Tam Date: Tue, 23 Oct 2018 17:33:40 +0700 Subject: [PATCH 5/5] move from SubscribeTx to Subscribe Special Tx in Double Validate --- core/genesis.go | 2 +- core/tx_pool.go | 2 +- eth/backend.go | 9 ++++++--- eth/downloader/api.go | 4 ++-- eth/handler_test.go | 8 ++++---- p2p/dial.go | 2 +- swarm/api/http/error.go | 2 +- 7 files changed, 16 insertions(+), 13 deletions(-) diff --git a/core/genesis.go b/core/genesis.go index 9d8eb27953..c87460b908 100644 --- a/core/genesis.go +++ b/core/genesis.go @@ -366,7 +366,7 @@ func DeveloperGenesisBlock(period uint64, faucet common.Address) *Genesis { common.BytesToAddress([]byte{6}): {Balance: big.NewInt(1)}, // ECAdd common.BytesToAddress([]byte{7}): {Balance: big.NewInt(1)}, // ECScalarMul common.BytesToAddress([]byte{8}): {Balance: big.NewInt(1)}, // ECPairing - faucet: {Balance: new(big.Int).Sub(new(big.Int).Lsh(big.NewInt(1), 256), big.NewInt(9))}, + faucet: {Balance: new(big.Int).Sub(new(big.Int).Lsh(big.NewInt(1), 256), big.NewInt(9))}, }, } } diff --git a/core/tx_pool.go b/core/tx_pool.go index e20aff5500..b07a924beb 100644 --- a/core/tx_pool.go +++ b/core/tx_pool.go @@ -81,7 +81,7 @@ var ( ErrZeroGasPrice = errors.New("zero gas price") - ErrDuplicateSpecialTransaction = errors.New("duplicate a specail transaction") + ErrDuplicateSpecialTransaction = errors.New("duplicate a special transaction") ) var ( diff --git a/eth/backend.go b/eth/backend.go index 4bea8f2d15..8213bf66fc 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -192,6 +192,9 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { // Hook double validation doubleValidateHook := func(block *types.Block) error { parentBlk := eth.blockchain.GetBlockByHash(block.ParentHash()) + if parentBlk == nil { + return fmt.Errorf("Fail to get parent block for hash: %v", block.ParentHash()) + } snap, err := c.GetSnapshot(eth.blockchain, parentBlk.Header()) if err != nil { if err == consensus.ErrUnknownAncestor { @@ -205,6 +208,9 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { return fmt.Errorf("Fail to validate M2 condition for importing block: %v", err) } if eth.etherbase != m2 { + txCh := make(chan core.TxPreEvent, txChanSize) + subEvent := eth.txPool.SubscribeSpecialTxPreEvent(txCh) + defer subEvent.Unsubscribe() // firstly, look into pending txPool pendingMap, err := eth.txPool.Pending() if err != nil { @@ -221,8 +227,6 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { } } //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) @@ -233,7 +237,6 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) { case <-time.After(time.Duration(10) * time.Second): return fmt.Errorf("Time out waiting for confirmation from m2") } - subEvent.Unsubscribe() } return nil } diff --git a/eth/downloader/api.go b/eth/downloader/api.go index d496fa6a4d..581e9aed24 100644 --- a/eth/downloader/api.go +++ b/eth/downloader/api.go @@ -40,8 +40,8 @@ type PublicDownloaderAPI struct { // installSyncSubscription channel. func NewPublicDownloaderAPI(d *Downloader, m *event.TypeMux) *PublicDownloaderAPI { api := &PublicDownloaderAPI{ - d: d, - mux: m, + d: d, + mux: m, installSyncSubscription: make(chan chan interface{}), uninstallSyncSubscription: make(chan *uninstallSyncSubscriptionRequest), } diff --git a/eth/handler_test.go b/eth/handler_test.go index e336dfa285..923df129a0 100644 --- a/eth/handler_test.go +++ b/eth/handler_test.go @@ -242,10 +242,10 @@ func testGetBlockBodies(t *testing.T, protocol int) { available []bool // Availability of explicitly requested blocks expected int // Total number of existing blocks to expect }{ - {1, nil, nil, 1}, // A single random block should be retrievable - {10, nil, nil, 10}, // Multiple random blocks should be retrievable - {limit, nil, nil, limit}, // The maximum possible blocks should be retrievable - {limit + 1, nil, nil, limit}, // No more than the possible block count should be returned + {1, nil, nil, 1}, // A single random block should be retrievable + {10, nil, nil, 10}, // Multiple random blocks should be retrievable + {limit, nil, nil, limit}, // The maximum possible blocks should be retrievable + {limit + 1, nil, nil, limit}, // No more than the possible block count should be returned {0, []common.Hash{pm.blockchain.Genesis().Hash()}, []bool{true}, 1}, // The genesis block should be retrievable {0, []common.Hash{pm.blockchain.CurrentBlock().Hash()}, []bool{true}, 1}, // The chains head block should be retrievable {0, []common.Hash{{}}, []bool{false}, 0}, // A non existent block should not be returned diff --git a/p2p/dial.go b/p2p/dial.go index 3a6cff18c1..f6ef432846 100644 --- a/p2p/dial.go +++ b/p2p/dial.go @@ -318,7 +318,7 @@ func (t *dialTask) Do(srv *Server) { } } if err == nil { - log.Trace("Dial pair connection sucess", "task", t.dest) + log.Trace("Dial pair connection success", "task", t.dest) } else { log.Trace("Dial pair connection error", "task", t.dest, "err", err) } diff --git a/swarm/api/http/error.go b/swarm/api/http/error.go index 9a65412cf9..2f77f2784a 100644 --- a/swarm/api/http/error.go +++ b/swarm/api/http/error.go @@ -71,7 +71,7 @@ func initErrHandling() { multipleChoicesPage := GetMultipleChoicesErrorPage() //map the codes to the available pages tnames := map[int]string{ - 0: genErrPage, //default + 0: genErrPage, //default http.StatusBadRequest: genErrPage, http.StatusNotFound: notFoundPage, http.StatusMultipleChoices: multipleChoicesPage,