From 56e1dc406d82da53738b6a6f59473137c7b01d05 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Wed, 19 Aug 2015 17:45:19 +0300 Subject: [PATCH] eth/fetcher: use announce numbers for distance checking --- eth/fetcher/fetcher.go | 68 ++++++++++++++++---------------- eth/fetcher/fetcher_test.go | 77 +++++++++++++++++++++++++++++-------- eth/handler.go | 2 +- 3 files changed, 96 insertions(+), 51 deletions(-) diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go index e528bd88d3..6c8d652269 100644 --- a/eth/fetcher/fetcher.go +++ b/eth/fetcher/fetcher.go @@ -93,10 +93,12 @@ type headerFilterTask struct { time time.Time // Arrival time of the headers } -// body represents the data contents (transactions and uncles) of a block. -type body struct { - transactions []*types.Transaction - uncles []*types.Header +// headerFilterTask represents a batch of block bodies (transactions and uncles) +// needing fetcher filtering. +type bodyFilterTask struct { + transactions [][]*types.Transaction // Collection of transactions per block bodies + uncles [][]*types.Header // Collection of uncles per block bodies + time time.Time // Arrival time of the blocks' contents } // inject represents a schedules import operation. @@ -114,7 +116,7 @@ type Fetcher struct { blockFilter chan chan []*types.Block headerFilter chan chan *headerFilterTask - bodyFilter chan chan []*body + bodyFilter chan chan *bodyFilterTask done chan common.Hash quit chan struct{} @@ -151,7 +153,7 @@ func New(getBlock blockRetrievalFn, validateBlock blockValidatorFn, broadcastBlo inject: make(chan *inject), blockFilter: make(chan chan []*types.Block), headerFilter: make(chan chan *headerFilterTask), - bodyFilter: make(chan chan []*body), + bodyFilter: make(chan chan *bodyFilterTask), done: make(chan common.Hash), quit: make(chan struct{}), announces: make(map[string]int), @@ -277,16 +279,11 @@ func (f *Fetcher) FilterHeaders(headers []*types.Header, time time.Time) []*type // FilterBodies extracts all the block bodies that were explicitly requested by // the fetcher, returning those that should be handled differently. -func (f *Fetcher) FilterBodies(transactions [][]*types.Transaction, uncles [][]*types.Header) ([][]*types.Transaction, [][]*types.Header) { +func (f *Fetcher) FilterBodies(transactions [][]*types.Transaction, uncles [][]*types.Header, time time.Time) ([][]*types.Transaction, [][]*types.Header) { glog.V(logger.Detail).Infof("[eth/62] filtering %d:%d bodies", len(transactions), len(uncles)) - // Assemble the body contents into a single data struct - bodies := make([]*body, 0, len(transactions)) - for i := 0; i < len(transactions) && i < len(uncles); i++ { - bodies = append(bodies, &body{transactions[i], uncles[i]}) - } // Send the filter channel to the fetcher - filter := make(chan []*body) + filter := make(chan *bodyFilterTask) select { case f.bodyFilter <- filter: @@ -295,19 +292,14 @@ func (f *Fetcher) FilterBodies(transactions [][]*types.Transaction, uncles [][]* } // Request the filtering of the body list select { - case filter <- bodies: + case filter <- &bodyFilterTask{transactions: transactions, uncles: uncles, time: time}: case <-f.quit: return nil, nil } // Retrieve the bodies remaining after filtering select { - case bodies := <-filter: - transactions, uncles = transactions[:0], uncles[:0] - for _, body := range bodies { - transactions = append(transactions, body.transactions) - uncles = append(uncles, body.uncles) - } - return transactions, uncles + case task := <-filter: + return task.transactions, task.uncles case <-f.quit: return nil, nil } @@ -361,6 +353,14 @@ func (f *Fetcher) loop() { glog.V(logger.Debug).Infof("Peer %s: exceeded outstanding announces (%d)", notification.origin, hashLimit) break } + // If we have a valid block number, check that it's potentially useful + if notification.number > 0 { + if dist := int64(notification.number) - int64(f.chainHeight()); dist < -maxUncleDist || dist > maxQueueDist { + glog.V(logger.Debug).Infof("[eth/62] Peer %s: discarded announcement #%d [%x], distance %d", notification.origin, notification.number, notification.hash[:4], dist) + discardMeter.Mark(1) + break + } + } // All is well, schedule the announce if block's not yet downloading if _, ok := f.fetching[notification.hash]; ok { break @@ -552,49 +552,51 @@ func (f *Fetcher) loop() { } case filter := <-f.bodyFilter: - // Block bodies arrived, extract any explicit completions, return all else - var bodies []*body + // Block bodies arrived, extract any explicitly requested blocks, return the rest + var task *bodyFilterTask select { - case bodies = <-filter: + case task = <-filter: case <-f.quit: return } - explicit, download := []*types.Block{}, []*body{} - for _, body := range bodies { + blocks := []*types.Block{} + for i := 0; i < len(task.transactions) && i < len(task.uncles); i++ { // Match up a body to any possible completion request matched := false for hash, announce := range f.completing { if f.queued[hash] == nil { - txnHash := types.DeriveSha(types.Transactions(body.transactions)) - uncleHash := types.CalcUncleHash(body.uncles) + txnHash := types.DeriveSha(types.Transactions(task.transactions[i])) + uncleHash := types.CalcUncleHash(task.uncles[i]) if txnHash == announce.header.TxHash && uncleHash == announce.header.UncleHash { // Mark the body matched, reassemble if still unknown matched = true if f.getBlock(hash) == nil { - explicit = append(explicit, types.NewBlockWithHeader(announce.header).WithBody(body.transactions, body.uncles)) + blocks = append(blocks, types.NewBlockWithHeader(announce.header).WithBody(task.transactions[i], task.uncles[i])) } else { f.forgetHash(hash) } } } } - if !matched { - download = append(download, body) + if matched { + task.transactions = append(task.transactions[:i], task.transactions[i+1:]...) + task.uncles = append(task.uncles[:i], task.uncles[i+1:]...) + i-- continue } } select { - case filter <- download: + case filter <- task: case <-f.quit: return } // Schedule the retrieved blocks for ordered import - for _, block := range explicit { + for _, block := range blocks { if announce := f.completing[block.Hash()]; announce != nil { f.enqueue(announce.origin, block) } diff --git a/eth/fetcher/fetcher_test.go b/eth/fetcher/fetcher_test.go index c798323052..dc7bdb7c93 100644 --- a/eth/fetcher/fetcher_test.go +++ b/eth/fetcher/fetcher_test.go @@ -169,7 +169,7 @@ func (f *fetcherTester) makeHeaderFetcher(blocks map[common.Hash]*types.Block, d } // makeBodyFetcher retrieves a block body fetcher associated with a simulated peer. -func (f *fetcherTester) makeBodyFetcher(blocks map[common.Hash]*types.Block) bodyRequesterFn { +func (f *fetcherTester) makeBodyFetcher(blocks map[common.Hash]*types.Block, drift time.Duration) bodyRequesterFn { closure := make(map[common.Hash]*types.Block) for hash, block := range blocks { closure[hash] = block @@ -187,7 +187,7 @@ func (f *fetcherTester) makeBodyFetcher(blocks map[common.Hash]*types.Block) bod } } // Return on a new thread - go f.fetcher.FilterBodies(transactions, uncles) + go f.fetcher.FilterBodies(transactions, uncles, time.Now().Add(drift)) return nil } @@ -242,7 +242,7 @@ func testConcurrentAnnouncements(t *testing.T, protocol int) { tester := newTester() blockFetcher := tester.makeBlockFetcher(blocks) headerFetcher := tester.makeHeaderFetcher(blocks, -gatherSlack) - bodyFetcher := tester.makeBodyFetcher(blocks) + bodyFetcher := tester.makeBodyFetcher(blocks, 0) counter := uint32(0) blockWrapper := func(hashes []common.Hash) error { @@ -292,7 +292,7 @@ func testOverlappingAnnouncements(t *testing.T, protocol int) { tester := newTester() blockFetcher := tester.makeBlockFetcher(blocks) headerFetcher := tester.makeHeaderFetcher(blocks, -gatherSlack) - bodyFetcher := tester.makeBodyFetcher(blocks) + bodyFetcher := tester.makeBodyFetcher(blocks, 0) // Iteratively announce blocks, but overlap them continuously fetching := make(chan []common.Hash) @@ -330,7 +330,7 @@ func testPendingDeduplication(t *testing.T, protocol int) { tester := newTester() blockFetcher := tester.makeBlockFetcher(blocks) headerFetcher := tester.makeHeaderFetcher(blocks, -gatherSlack) - bodyFetcher := tester.makeBodyFetcher(blocks) + bodyFetcher := tester.makeBodyFetcher(blocks, 0) delay := 50 * time.Millisecond counter := uint32(0) @@ -390,7 +390,7 @@ func testRandomArrivalImport(t *testing.T, protocol int) { tester := newTester() blockFetcher := tester.makeBlockFetcher(blocks) headerFetcher := tester.makeHeaderFetcher(blocks, -gatherSlack) - bodyFetcher := tester.makeBodyFetcher(blocks) + bodyFetcher := tester.makeBodyFetcher(blocks, 0) // Iteratively announce blocks, skipping one entry imported := make(chan *types.Block, len(hashes)-1) @@ -431,7 +431,7 @@ func testQueueGapFill(t *testing.T, protocol int) { tester := newTester() blockFetcher := tester.makeBlockFetcher(blocks) headerFetcher := tester.makeHeaderFetcher(blocks, -gatherSlack) - bodyFetcher := tester.makeBodyFetcher(blocks) + bodyFetcher := tester.makeBodyFetcher(blocks, 0) // Iteratively announce blocks, skipping one entry imported := make(chan *types.Block, len(hashes)-1) @@ -467,7 +467,7 @@ func testImportDeduplication(t *testing.T, protocol int) { tester := newTester() blockFetcher := tester.makeBlockFetcher(blocks) headerFetcher := tester.makeHeaderFetcher(blocks, -gatherSlack) - bodyFetcher := tester.makeBodyFetcher(blocks) + bodyFetcher := tester.makeBodyFetcher(blocks, 0) counter := uint32(0) tester.fetcher.insertChain = func(blocks types.Blocks) (int, error) { @@ -502,31 +502,74 @@ func testImportDeduplication(t *testing.T, protocol int) { } // Tests that blocks with numbers much lower or higher than out current head get -// discarded no prevent wasting resources on useless blocks from faulty peers. -func TestDistantDiscarding(t *testing.T) { - // Create a long chain to import +// discarded to prevent wasting resources on useless blocks from faulty peers. +func TestDistantPropagationDiscarding(t *testing.T) { + // Create a long chain to import and define the discard boundaries hashes, blocks := makeChain(3*maxQueueDist, 0, genesis) head := hashes[len(hashes)/2] + low, high := len(hashes)/2+maxUncleDist+1, len(hashes)/2-maxQueueDist-1 + // Create a tester and simulate a head block being the middle of the above chain tester := newTester() tester.hashes = []common.Hash{head} tester.blocks = map[common.Hash]*types.Block{head: blocks[head]} // Ensure that a block with a lower number than the threshold is discarded - tester.fetcher.Enqueue("lower", blocks[hashes[0]]) + tester.fetcher.Enqueue("lower", blocks[hashes[low]]) time.Sleep(10 * time.Millisecond) if !tester.fetcher.queue.Empty() { t.Fatalf("fetcher queued stale block") } // Ensure that a block with a higher number than the threshold is discarded - tester.fetcher.Enqueue("higher", blocks[hashes[len(hashes)-1]]) + tester.fetcher.Enqueue("higher", blocks[hashes[high]]) time.Sleep(10 * time.Millisecond) if !tester.fetcher.queue.Empty() { t.Fatalf("fetcher queued future block") } } +// Tests that announcements with numbers much lower or higher than out current +// head get discarded to prevent wasting resources on useless blocks from faulty +// peers. +func TestDistantAnnouncementDiscarding62(t *testing.T) { testDistantAnnouncementDiscarding(t, 62) } +func TestDistantAnnouncementDiscarding63(t *testing.T) { testDistantAnnouncementDiscarding(t, 63) } +func TestDistantAnnouncementDiscarding64(t *testing.T) { testDistantAnnouncementDiscarding(t, 64) } + +func testDistantAnnouncementDiscarding(t *testing.T, protocol int) { + // Create a long chain to import and define the discard boundaries + hashes, blocks := makeChain(3*maxQueueDist, 0, genesis) + head := hashes[len(hashes)/2] + + low, high := len(hashes)/2+maxUncleDist+1, len(hashes)/2-maxQueueDist-1 + + // Create a tester and simulate a head block being the middle of the above chain + tester := newTester() + tester.hashes = []common.Hash{head} + tester.blocks = map[common.Hash]*types.Block{head: blocks[head]} + + headerFetcher := tester.makeHeaderFetcher(blocks, -gatherSlack) + bodyFetcher := tester.makeBodyFetcher(blocks, 0) + + fetching := make(chan struct{}, 2) + tester.fetcher.fetchingHook = func(hashes []common.Hash) { fetching <- struct{}{} } + + // Ensure that a block with a lower number than the threshold is discarded + tester.fetcher.Notify("lower", hashes[low], blocks[hashes[low]].NumberU64(), time.Now().Add(-arriveTimeout), nil, headerFetcher, bodyFetcher) + select { + case <-time.After(50 * time.Millisecond): + case <-fetching: + t.Fatalf("fetcher requested stale header") + } + // Ensure that a block with a higher number than the threshold is discarded + tester.fetcher.Notify("higher", hashes[high], blocks[hashes[high]].NumberU64(), time.Now().Add(-arriveTimeout), nil, headerFetcher, bodyFetcher) + select { + case <-time.After(50 * time.Millisecond): + case <-fetching: + t.Fatalf("fetcher requested future header") + } +} + // Tests that a peer is unable to use unbounded memory with sending infinite // block announcements to a node, but that even in the face of such an attack, // the fetcher remains operational. @@ -547,12 +590,12 @@ func testHashMemoryExhaustionAttack(t *testing.T, protocol int) { hashes, blocks := makeChain(targetBlocks, 0, genesis) validBlockFetcher := tester.makeBlockFetcher(blocks) validHeaderFetcher := tester.makeHeaderFetcher(blocks, -gatherSlack) - validBodyFetcher := tester.makeBodyFetcher(blocks) + validBodyFetcher := tester.makeBodyFetcher(blocks, 0) attack, _ := makeChain(targetBlocks, 0, unknownBlock) attackerBlockFetcher := tester.makeBlockFetcher(nil) attackerHeaderFetcher := tester.makeHeaderFetcher(nil, -gatherSlack) - attackerBodyFetcher := tester.makeBodyFetcher(nil) + attackerBodyFetcher := tester.makeBodyFetcher(nil, 0) // Feed the tester a huge hashset from the attacker, and a limited from the valid peer for i := 0; i < len(attack); i++ { @@ -566,7 +609,7 @@ func testHashMemoryExhaustionAttack(t *testing.T, protocol int) { if protocol < 62 { tester.fetcher.Notify("attacker", attack[i], 0, time.Now(), attackerBlockFetcher, nil, nil) } else { - tester.fetcher.Notify("attacker", attack[i], uint64(i+1), time.Now(), nil, attackerHeaderFetcher, attackerBodyFetcher) + tester.fetcher.Notify("attacker", attack[i], 1 /* don't distance drop */, time.Now(), nil, attackerHeaderFetcher, attackerBodyFetcher) } } if len(tester.fetcher.announced) != hashLimit+maxQueueDist { @@ -580,7 +623,7 @@ func testHashMemoryExhaustionAttack(t *testing.T, protocol int) { if protocol < 62 { tester.fetcher.Notify("valid", hashes[i], 0, time.Now().Add(-arriveTimeout), validBlockFetcher, nil, nil) } else { - tester.fetcher.Notify("valid", hashes[i], uint64(i+1), time.Now().Add(-arriveTimeout), nil, validHeaderFetcher, validBodyFetcher) + tester.fetcher.Notify("valid", hashes[i], uint64(len(hashes)-i-1), time.Now().Add(-arriveTimeout), nil, validHeaderFetcher, validBodyFetcher) } verifyImportEvent(t, imported) } diff --git a/eth/handler.go b/eth/handler.go index 3d8d07ed54..e7404e36a7 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -436,7 +436,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { uncles[i] = body.Uncles } // Filter out any explicitly requested bodies, deliver the rest to the downloader - if trasactions, uncles := pm.fetcher.FilterBodies(trasactions, uncles); len(trasactions) > 0 || len(uncles) > 0 { + if trasactions, uncles := pm.fetcher.FilterBodies(trasactions, uncles, time.Now()); len(trasactions) > 0 || len(uncles) > 0 { err := pm.downloader.DeliverBodies(p.id, trasactions, uncles) if err != nil { glog.V(logger.Debug).Infoln(err)