From d36c25bcbcbc1986ab238465a187789c82818238 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Mon, 22 Jun 2015 14:07:08 +0300 Subject: [PATCH 1/5] eth/fetcher: handle and test block announce DOS attacks --- eth/fetcher/fetcher.go | 46 +++++++++++++++++++++++++++++++------ eth/fetcher/fetcher_test.go | 43 ++++++++++++++++++++++++++++++++++ 2 files changed, 82 insertions(+), 7 deletions(-) diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go index b80182a455..8d707da5c4 100644 --- a/eth/fetcher/fetcher.go +++ b/eth/fetcher/fetcher.go @@ -20,6 +20,7 @@ const ( fetchTimeout = 5 * time.Second // Maximum alloted time to return an explicitly requested block maxUncleDist = 7 // Maximum allowed backward distance from the chain head maxQueueDist = 32 // Maximum allowed distance from the chain head to queue + announceLimit = 256 // Maximum number of unique blocks a peer may have announced ) var ( @@ -74,6 +75,7 @@ type Fetcher struct { quit chan struct{} // Announce states + announces map[string]int // Per peer announce counts to prevent memory exhaustion announced map[common.Hash][]*announce // Announced blocks, scheduled for fetching fetching map[common.Hash]*announce // Announced blocks, currently fetching @@ -98,6 +100,7 @@ func New(getBlock blockRetrievalFn, validateBlock blockValidatorFn, broadcastBlo filter: make(chan chan []*types.Block), done: make(chan common.Hash), quit: make(chan struct{}), + announces: make(map[string]int), announced: make(map[common.Hash][]*announce), fetching: make(map[common.Hash]*announce), queue: prque.New(), @@ -189,8 +192,7 @@ func (f *Fetcher) loop() { // Clean up any expired block fetches for hash, announce := range f.fetching { if time.Since(announce.time) > fetchTimeout { - delete(f.announced, hash) - delete(f.fetching, hash) + f.forgetBlock(hash) } } // Import any queued blocks that could potentially fit @@ -217,10 +219,17 @@ func (f *Fetcher) loop() { return case notification := <-f.notify: - // A block was announced, schedule if it's not yet downloading + // A block was announced, make sure the peer isn't DOSing us + count := f.announces[notification.origin] + 1 + if count > announceLimit { + glog.V(logger.Debug).Infof("Peer %s: exceeded outstanding announces (%d)", notification.origin, announceLimit) + break + } + // All is well, schedule the announce if block's not yet downloading if _, ok := f.fetching[notification.hash]; ok { break } + f.announces[notification.origin] = count f.announced[notification.hash] = append(f.announced[notification.hash], notification) if len(f.announced) == 1 { f.reschedule(fetch) @@ -232,8 +241,7 @@ func (f *Fetcher) loop() { case hash := <-f.done: // A pending import finished, remove all traces of the notification - delete(f.announced, hash) - delete(f.fetching, hash) + f.forgetBlock(hash) delete(f.queued, hash) case <-fetch.C: @@ -242,12 +250,15 @@ func (f *Fetcher) loop() { for hash, announces := range f.announced { if time.Since(announces[0].time) > arriveTimeout-gatherSlack { + // Pick a random peer to retrieve from, reset all others announce := announces[rand.Intn(len(announces))] + f.forgetBlock(hash) + + // If the block still didn't arrive, queue for fetching if f.getBlock(hash) == nil { request[announce.origin] = append(request[announce.origin], hash) f.fetching[hash] = announce } - delete(f.announced, hash) } } // Send out all block requests @@ -285,7 +296,7 @@ func (f *Fetcher) loop() { if f.getBlock(hash) == nil { explicit = append(explicit, block) } else { - delete(f.fetching, hash) + f.forgetBlock(hash) } } else { download = append(download, block) @@ -377,3 +388,24 @@ func (f *Fetcher) insert(peer string, block *types.Block) { go f.broadcastBlock(block, false) }() } + +// forgetBlock removes all traces of a block from the fetcher's internal state. +func (f *Fetcher) forgetBlock(hash common.Hash) { + // Remove all pending announces and decrement DOS counters + for _, announce := range f.announced[hash] { + f.announces[announce.origin]-- + if f.announces[announce.origin] == 0 { + delete(f.announces, announce.origin) + } + } + delete(f.announced, hash) + + // Remove any pending fetches and decrement the DOS counters + if announce := f.fetching[hash]; announce != nil { + f.announces[announce.origin]-- + if f.announces[announce.origin] == 0 { + delete(f.announces, announce.origin) + } + delete(f.fetching, hash) + } +} diff --git a/eth/fetcher/fetcher_test.go b/eth/fetcher/fetcher_test.go index 0d069ac65e..d594d830c8 100644 --- a/eth/fetcher/fetcher_test.go +++ b/eth/fetcher/fetcher_test.go @@ -395,3 +395,46 @@ func TestDistantDiscarding(t *testing.T) { t.Fatalf("fetcher queued future block") } } + +// 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. +func TestAnnounceMemoryExhaustionAttack(t *testing.T) { + tester := newTester() + + // Create a valid chain and an infinite junk chain + hashes := createHashes(announceLimit+2*maxQueueDist, knownHash) + blocks := createBlocksFromHashes(hashes) + valid := tester.makeFetcher(blocks) + + attack := createHashes(announceLimit+2*maxQueueDist, unknownHash) + attacker := tester.makeFetcher(nil) + + // Feed the tester a huge hashset from the attacker, and a limited from the valid peer + for i := 0; i < len(attack); i++ { + if i < maxQueueDist { + tester.fetcher.Notify("valid", hashes[len(hashes)-1-i], time.Now().Add(arriveTimeout/2), valid) + } + tester.fetcher.Notify("attacker", attack[i], time.Now().Add(arriveTimeout/2), attacker) + } + if len(tester.fetcher.announced) != announceLimit+maxQueueDist { + t.Fatalf("queued announce count mismatch: have %d, want %d", len(tester.fetcher.announced), announceLimit+maxQueueDist) + } + // Wait for synchronisation to complete and check success for the valid peer + time.Sleep(2 * arriveTimeout) + if imported := len(tester.blocks); imported != maxQueueDist { + t.Fatalf("partial synchronised block mismatch: have %v, want %v", imported, maxQueueDist) + } + // Feed the remaining valid hashes to ensure DOS protection state remains clean + for i := len(hashes) - maxQueueDist; i >= 0; { + for j := 0; j < maxQueueDist && i >= 0; j++ { + tester.fetcher.Notify("valid", hashes[i], time.Now().Add(time.Millisecond), valid) + i-- + } + time.Sleep(256 * time.Millisecond) + } + time.Sleep(256 * time.Millisecond) + if imported := len(tester.blocks); imported != len(hashes) { + t.Fatalf("fully synchronised block mismatch: have %v, want %v", imported, len(hashes)) + } +} From 1989d1491a21ec1dd8adb20906e07badc5a2f9c9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Mon, 22 Jun 2015 16:49:47 +0300 Subject: [PATCH 2/5] eth/fetcher: handle and (crude) test block memory DOS --- eth/fetcher/fetcher.go | 61 ++++++++++++++++++++++++--------- eth/fetcher/fetcher_test.go | 67 +++++++++++++++++++++++++++++++++---- 2 files changed, 105 insertions(+), 23 deletions(-) diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go index 8d707da5c4..ceca79df0f 100644 --- a/eth/fetcher/fetcher.go +++ b/eth/fetcher/fetcher.go @@ -20,7 +20,8 @@ const ( fetchTimeout = 5 * time.Second // Maximum alloted time to return an explicitly requested block maxUncleDist = 7 // Maximum allowed backward distance from the chain head maxQueueDist = 32 // Maximum allowed distance from the chain head to queue - announceLimit = 256 // Maximum number of unique blocks a peer may have announced + hashLimit = 256 // Maximum number of unique blocks a peer may have announced + blockLimit = 64 // Maximum number of unique blocks a per may have delivered ) var ( @@ -80,8 +81,9 @@ type Fetcher struct { fetching map[common.Hash]*announce // Announced blocks, currently fetching // Block cache - queue *prque.Prque // Queue containing the import operations (block number sorted) - queued map[common.Hash]struct{} // Presence set of already queued blocks (to dedup imports) + queue *prque.Prque // Queue containing the import operations (block number sorted) + queues map[string]int // Per peer block counts to prevent memory exhaustion + queued map[common.Hash]*inject // Set of already queued blocks (to dedup imports) // Callbacks getBlock blockRetrievalFn // Retrieves a block from the local chain @@ -104,7 +106,8 @@ func New(getBlock blockRetrievalFn, validateBlock blockValidatorFn, broadcastBlo announced: make(map[common.Hash][]*announce), fetching: make(map[common.Hash]*announce), queue: prque.New(), - queued: make(map[common.Hash]struct{}), + queues: make(map[string]int), + queued: make(map[common.Hash]*inject), getBlock: getBlock, validateBlock: validateBlock, broadcastBlock: broadcastBlock, @@ -192,22 +195,24 @@ func (f *Fetcher) loop() { // Clean up any expired block fetches for hash, announce := range f.fetching { if time.Since(announce.time) > fetchTimeout { - f.forgetBlock(hash) + f.forgetHash(hash) } } // Import any queued blocks that could potentially fit height := f.chainHeight() for !f.queue.Empty() { op := f.queue.PopItem().(*inject) - number := op.block.NumberU64() // If too high up the chain or phase, continue later + number := op.block.NumberU64() if number > height+1 { f.queue.Push(op, -float32(op.block.NumberU64())) break } // Otherwise if fresh and still unknown, try and import - if number+maxUncleDist < height || f.getBlock(op.block.Hash()) != nil { + hash := op.block.Hash() + if number+maxUncleDist < height || f.getBlock(hash) != nil { + f.forgetBlock(hash) continue } f.insert(op.origin, op.block) @@ -221,8 +226,8 @@ func (f *Fetcher) loop() { case notification := <-f.notify: // A block was announced, make sure the peer isn't DOSing us count := f.announces[notification.origin] + 1 - if count > announceLimit { - glog.V(logger.Debug).Infof("Peer %s: exceeded outstanding announces (%d)", notification.origin, announceLimit) + if count > hashLimit { + glog.V(logger.Debug).Infof("Peer %s: exceeded outstanding announces (%d)", notification.origin, hashLimit) break } // All is well, schedule the announce if block's not yet downloading @@ -241,8 +246,8 @@ func (f *Fetcher) loop() { case hash := <-f.done: // A pending import finished, remove all traces of the notification + f.forgetHash(hash) f.forgetBlock(hash) - delete(f.queued, hash) case <-fetch.C: // At least one block's timer ran out, check for needing retrieval @@ -252,7 +257,7 @@ func (f *Fetcher) loop() { if time.Since(announces[0].time) > arriveTimeout-gatherSlack { // Pick a random peer to retrieve from, reset all others announce := announces[rand.Intn(len(announces))] - f.forgetBlock(hash) + f.forgetHash(hash) // If the block still didn't arrive, queue for fetching if f.getBlock(hash) == nil { @@ -296,7 +301,7 @@ func (f *Fetcher) loop() { if f.getBlock(hash) == nil { explicit = append(explicit, block) } else { - f.forgetBlock(hash) + f.forgetHash(hash) } } else { download = append(download, block) @@ -339,6 +344,12 @@ func (f *Fetcher) reschedule(fetch *time.Timer) { func (f *Fetcher) enqueue(peer string, block *types.Block) { hash := block.Hash() + // Ensure the peer isn't DOSing us + count := f.queues[peer] + 1 + if count > blockLimit { + glog.V(logger.Debug).Infof("Peer %s: discarded block #%d [%x], exceeded allowance (%d)", peer, block.NumberU64(), hash.Bytes()[:4], blockLimit) + return + } // Discard any past or too distant blocks if dist := int64(block.NumberU64()) - int64(f.chainHeight()); dist < -maxUncleDist || dist > maxQueueDist { glog.V(logger.Debug).Infof("Peer %s: discarded block #%d [%x], distance %d", peer, block.NumberU64(), hash.Bytes()[:4], dist) @@ -346,8 +357,13 @@ func (f *Fetcher) enqueue(peer string, block *types.Block) { } // Schedule the block for future importing if _, ok := f.queued[hash]; !ok { - f.queued[hash] = struct{}{} - f.queue.Push(&inject{origin: peer, block: block}, -float32(block.NumberU64())) + op := &inject{ + origin: peer, + block: block, + } + f.queues[peer] = count + f.queued[hash] = op + f.queue.Push(op, -float32(block.NumberU64())) if glog.V(logger.Debug) { glog.Infof("Peer %s: queued block #%d [%x], total %v", peer, block.NumberU64(), hash.Bytes()[:4], f.queue.Size()) @@ -389,8 +405,9 @@ func (f *Fetcher) insert(peer string, block *types.Block) { }() } -// forgetBlock removes all traces of a block from the fetcher's internal state. -func (f *Fetcher) forgetBlock(hash common.Hash) { +// forgetHash removes all traces of a block announcement from the fetcher's +// internal state. +func (f *Fetcher) forgetHash(hash common.Hash) { // Remove all pending announces and decrement DOS counters for _, announce := range f.announced[hash] { f.announces[announce.origin]-- @@ -409,3 +426,15 @@ func (f *Fetcher) forgetBlock(hash common.Hash) { delete(f.fetching, hash) } } + +// forgetBlock removes all traces of a queued block frmo the fetcher's internal +// state. +func (f *Fetcher) forgetBlock(hash common.Hash) { + if insert := f.queued[hash]; insert != nil { + f.queues[insert.origin]-- + if f.queues[insert.origin] == 0 { + delete(f.queues, insert.origin) + } + delete(f.queued, hash) + } +} diff --git a/eth/fetcher/fetcher_test.go b/eth/fetcher/fetcher_test.go index d594d830c8..b9f0f36a51 100644 --- a/eth/fetcher/fetcher_test.go +++ b/eth/fetcher/fetcher_test.go @@ -399,15 +399,15 @@ func TestDistantDiscarding(t *testing.T) { // 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. -func TestAnnounceMemoryExhaustionAttack(t *testing.T) { +func TestHashMemoryExhaustionAttack(t *testing.T) { tester := newTester() // Create a valid chain and an infinite junk chain - hashes := createHashes(announceLimit+2*maxQueueDist, knownHash) + hashes := createHashes(hashLimit+2*maxQueueDist, knownHash) blocks := createBlocksFromHashes(hashes) valid := tester.makeFetcher(blocks) - attack := createHashes(announceLimit+2*maxQueueDist, unknownHash) + attack := createHashes(hashLimit+2*maxQueueDist, unknownHash) attacker := tester.makeFetcher(nil) // Feed the tester a huge hashset from the attacker, and a limited from the valid peer @@ -417,8 +417,8 @@ func TestAnnounceMemoryExhaustionAttack(t *testing.T) { } tester.fetcher.Notify("attacker", attack[i], time.Now().Add(arriveTimeout/2), attacker) } - if len(tester.fetcher.announced) != announceLimit+maxQueueDist { - t.Fatalf("queued announce count mismatch: have %d, want %d", len(tester.fetcher.announced), announceLimit+maxQueueDist) + if len(tester.fetcher.announced) != hashLimit+maxQueueDist { + t.Fatalf("queued announce count mismatch: have %d, want %d", len(tester.fetcher.announced), hashLimit+maxQueueDist) } // Wait for synchronisation to complete and check success for the valid peer time.Sleep(2 * arriveTimeout) @@ -431,10 +431,63 @@ func TestAnnounceMemoryExhaustionAttack(t *testing.T) { tester.fetcher.Notify("valid", hashes[i], time.Now().Add(time.Millisecond), valid) i-- } - time.Sleep(256 * time.Millisecond) + time.Sleep(500 * time.Millisecond) } - time.Sleep(256 * time.Millisecond) + time.Sleep(500 * time.Millisecond) if imported := len(tester.blocks); imported != len(hashes) { t.Fatalf("fully synchronised block mismatch: have %v, want %v", imported, len(hashes)) } } + +// Tests that blocks sent to the fetcher (either through propagation or via hash +// announces and retrievals) don't pile up indefinitely, exhausting available +// system memory. +func TestBlockMemoryExhaustionAttack(t *testing.T) { + tester := newTester() + + // Create a valid chain and a batch of dangling (but in range) blocks + hashes := createHashes(blockLimit, knownHash) + blocks := createBlocksFromHashes(hashes) + + attack := make(map[common.Hash]*types.Block) + for i := 0; i < 16; i++ { + hashes := createHashes(maxQueueDist-1, unknownHash) + blocks := createBlocksFromHashes(hashes) + for _, hash := range hashes[:maxQueueDist-2] { + attack[hash] = blocks[hash] + } + } + // Try to feed all the attacker blocks make sure only a limited batch is accepted + for _, block := range attack { + tester.fetcher.Enqueue("attacker", block) + } + time.Sleep(100 * time.Millisecond) + if queued := tester.fetcher.queue.Size(); queued != blockLimit { + t.Fatalf("queued block count mismatch: have %d, want %d", queued, blockLimit) + } + // Queue up a batch of valid blocks, and check that a new peer is allowed to do so + for i := 0; i < maxQueueDist-1; i++ { + tester.fetcher.Enqueue("valid", blocks[hashes[len(hashes)-3-i]]) + } + time.Sleep(100 * time.Millisecond) + if queued := tester.fetcher.queue.Size(); queued != blockLimit+maxQueueDist-1 { + t.Fatalf("queued block count mismatch: have %d, want %d", queued, blockLimit+maxQueueDist-1) + } + // Insert the missing piece (and sanity check the import) + tester.fetcher.Enqueue("valid", blocks[hashes[len(hashes)-2]]) + time.Sleep(500 * time.Millisecond) + if imported := len(tester.blocks); imported != maxQueueDist+1 { + t.Fatalf("synchronised block mismatch: have %v, want %v", imported, maxQueueDist+1) + } + // Insert the remaining blocks in chunks to ensure clean DOS protection + for i := maxQueueDist; i < len(hashes)-1; i++ { + tester.fetcher.Enqueue("valid", blocks[hashes[len(hashes)-2-i]]) + if i%maxQueueDist == 0 { + time.Sleep(500 * time.Millisecond) + } + } + time.Sleep(500 * time.Millisecond) + if imported := len(tester.blocks); imported != len(hashes) { + t.Fatalf("synchronised block mismatch: have %v, want %v", imported, len(hashes)) + } +} From b53f701c2791e7becba74c1efd4800fe68a06707 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Mon, 22 Jun 2015 18:08:28 +0300 Subject: [PATCH 3/5] eth/fetcher: remove test sleeps (15s -> 2.8s) --- eth/fetcher/fetcher.go | 17 ++- eth/fetcher/fetcher_test.go | 209 ++++++++++++++++++++++++++---------- 2 files changed, 167 insertions(+), 59 deletions(-) diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go index ceca79df0f..7b5804ab22 100644 --- a/eth/fetcher/fetcher.go +++ b/eth/fetcher/fetcher.go @@ -92,6 +92,10 @@ type Fetcher struct { chainHeight chainHeightFn // Retrieves the current chain's height insertChain chainInsertFn // Injects a batch of blocks into the chain dropPeer peerDropFn // Drops a peer for misbehaving + + // Testing hooks + fetchingHook func([]common.Hash) // Method to call upon starting a block fetch + importedHook func(*types.Block) // Method to call upon successful block import } // New creates a block fetcher to retrieve blocks based on hash announcements. @@ -277,7 +281,13 @@ func (f *Fetcher) loop() { glog.V(logger.Detail).Infof("Peer %s: fetching %s", peer, list) } - go f.fetching[hashes[0]].fetch(hashes) + hashes := hashes // closure! + go func() { + if f.fetchingHook != nil { + f.fetchingHook(hashes) + } + f.fetching[hashes[0]].fetch(hashes) + }() } // Schedule the next fetch if blocks are still pending f.reschedule(fetch) @@ -402,6 +412,11 @@ func (f *Fetcher) insert(peer string, block *types.Block) { } // If import succeeded, broadcast the block go f.broadcastBlock(block, false) + + // Invoke the testing hook if needed + if f.importedHook != nil { + f.importedHook(block) + } }() } diff --git a/eth/fetcher/fetcher_test.go b/eth/fetcher/fetcher_test.go index b9f0f36a51..06ca2ef86f 100644 --- a/eth/fetcher/fetcher_test.go +++ b/eth/fetcher/fetcher_test.go @@ -163,7 +163,7 @@ func (f *fetcherTester) makeFetcher(blocks map[common.Hash]*types.Block) blockRe // them, successfully importing into the local chain. func TestSequentialAnnouncements(t *testing.T) { // Create a chain of blocks to import - targetBlocks := 24 + targetBlocks := 4 * hashLimit hashes := createHashes(targetBlocks, knownHash) blocks := createBlocksFromHashes(hashes) @@ -171,12 +171,22 @@ func TestSequentialAnnouncements(t *testing.T) { fetcher := tester.makeFetcher(blocks) // Iteratively announce blocks until all are imported - for i := len(hashes) - 1; i >= 0; i-- { + imported := make(chan *types.Block) + tester.fetcher.importedHook = func(block *types.Block) { imported <- block } + + for i := len(hashes) - 2; i >= 0; i-- { tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher) - time.Sleep(50 * time.Millisecond) + + select { + case <-imported: + case <-time.After(time.Second): + t.Fatalf("block %d: import timeout", len(hashes)-i) + } } - if imported := len(tester.blocks); imported != targetBlocks+1 { - t.Fatalf("synchronised block mismatch: have %v, want %v", imported, targetBlocks+1) + select { + case <-imported: + t.Fatalf("extra block imported") + case <-time.After(50 * time.Millisecond): } } @@ -184,7 +194,7 @@ func TestSequentialAnnouncements(t *testing.T) { // peer), they will only get downloaded at most once. func TestConcurrentAnnouncements(t *testing.T) { // Create a chain of blocks to import - targetBlocks := 24 + targetBlocks := 4 * hashLimit hashes := createHashes(targetBlocks, knownHash) blocks := createBlocksFromHashes(hashes) @@ -198,15 +208,24 @@ func TestConcurrentAnnouncements(t *testing.T) { return fetcher(hashes) } // Iteratively announce blocks until all are imported - for i := len(hashes) - 1; i >= 0; i-- { + imported := make(chan *types.Block) + tester.fetcher.importedHook = func(block *types.Block) { imported <- block } + + for i := len(hashes) - 2; i >= 0; i-- { tester.fetcher.Notify("first", hashes[i], time.Now().Add(-arriveTimeout), wrapper) tester.fetcher.Notify("second", hashes[i], time.Now().Add(-arriveTimeout+time.Millisecond), wrapper) tester.fetcher.Notify("second", hashes[i], time.Now().Add(-arriveTimeout-time.Millisecond), wrapper) - time.Sleep(50 * time.Millisecond) + select { + case <-imported: + case <-time.After(time.Second): + t.Fatalf("block %d: import timeout", len(hashes)-i) + } } - if imported := len(tester.blocks); imported != targetBlocks+1 { - t.Fatalf("synchronised block mismatch: have %v, want %v", imported, targetBlocks+1) + select { + case <-imported: + t.Fatalf("extra block imported") + case <-time.After(50 * time.Millisecond): } // Make sure no blocks were retrieved twice if int(counter) != targetBlocks { @@ -218,7 +237,7 @@ func TestConcurrentAnnouncements(t *testing.T) { // results in a valid import. func TestOverlappingAnnouncements(t *testing.T) { // Create a chain of blocks to import - targetBlocks := 24 + targetBlocks := 4 * hashLimit hashes := createHashes(targetBlocks, knownHash) blocks := createBlocksFromHashes(hashes) @@ -226,15 +245,31 @@ func TestOverlappingAnnouncements(t *testing.T) { fetcher := tester.makeFetcher(blocks) // Iteratively announce blocks, but overlap them continuously - delay, overlap := 50*time.Millisecond, time.Duration(5) - for i := len(hashes) - 1; i >= 0; i-- { - tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout+overlap*delay), fetcher) - time.Sleep(delay) - } - time.Sleep(overlap * delay) + 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 } - if imported := len(tester.blocks); imported != targetBlocks+1 { - t.Fatalf("synchronised block mismatch: have %v, want %v", imported, targetBlocks+1) + for i := len(hashes) - 2; i >= 0; i-- { + tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher) + select { + case <-fetching: + case <-time.After(time.Second): + t.Fatalf("hash %d: announce timeout", len(hashes)-i) + } + } + // Wait for all the imports to complete and check count + for i := 0; i < len(hashes)-1; i++ { + select { + case <-imported: + case <-time.After(time.Second): + t.Fatalf("block %d: import timeout", i) + } + } + select { + case <-imported: + t.Fatalf("extra block imported") + case <-time.After(50 * time.Millisecond): } } @@ -280,27 +315,37 @@ func TestPendingDeduplication(t *testing.T) { // imported when all the gaps are filled in. func TestRandomArrivalImport(t *testing.T) { // Create a chain of blocks to import, and choose one to delay - targetBlocks := 24 - hashes := createHashes(targetBlocks, knownHash) + hashes := createHashes(maxQueueDist, knownHash) blocks := createBlocksFromHashes(hashes) - skip := targetBlocks / 2 + skip := maxQueueDist / 2 tester := newTester() fetcher := tester.makeFetcher(blocks) // Iteratively announce blocks, skipping one entry + imported := make(chan *types.Block, len(hashes)-1) + tester.fetcher.importedHook = func(block *types.Block) { imported <- block } + for i := len(hashes) - 1; i >= 0; i-- { if i != skip { tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher) - time.Sleep(50 * time.Millisecond) + time.Sleep(time.Millisecond) } } // Finally announce the skipped entry and check full import tester.fetcher.Notify("valid", hashes[skip], time.Now().Add(-arriveTimeout), fetcher) - time.Sleep(50 * time.Millisecond) - if imported := len(tester.blocks); imported != targetBlocks+1 { - t.Fatalf("synchronised block mismatch: have %v, want %v", imported, targetBlocks+1) + for i := 0; i < len(hashes)-1; i++ { + select { + case <-imported: + case <-time.After(time.Second): + t.Fatalf("block %d: import timeout", i) + } + } + select { + case <-imported: + t.Fatalf("extra block imported") + case <-time.After(50 * time.Millisecond): } } @@ -308,27 +353,37 @@ func TestRandomArrivalImport(t *testing.T) { // are correctly schedule, filling and import queue gaps. func TestQueueGapFill(t *testing.T) { // Create a chain of blocks to import, and choose one to not announce at all - targetBlocks := 24 - hashes := createHashes(targetBlocks, knownHash) + hashes := createHashes(maxQueueDist, knownHash) blocks := createBlocksFromHashes(hashes) - skip := targetBlocks / 2 + skip := maxQueueDist / 2 tester := newTester() fetcher := tester.makeFetcher(blocks) // Iteratively announce blocks, skipping one entry + imported := make(chan *types.Block, len(hashes)-1) + tester.fetcher.importedHook = func(block *types.Block) { imported <- block } + for i := len(hashes) - 1; i >= 0; i-- { if i != skip { tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher) - time.Sleep(50 * time.Millisecond) + time.Sleep(time.Millisecond) } } // Fill the missing block directly as if propagated tester.fetcher.Enqueue("valid", blocks[hashes[skip]]) - time.Sleep(50 * time.Millisecond) - if imported := len(tester.blocks); imported != targetBlocks+1 { - t.Fatalf("synchronised block mismatch: have %v, want %v", imported, targetBlocks+1) + for i := 0; i < len(hashes)-1; i++ { + select { + case <-imported: + case <-time.After(time.Second): + t.Fatalf("block %d: import timeout", i) + } + } + select { + case <-imported: + t.Fatalf("extra block imported") + case <-time.After(50 * time.Millisecond): } } @@ -348,9 +403,15 @@ func TestImportDeduplication(t *testing.T) { atomic.AddUint32(&counter, uint32(len(blocks))) return tester.insertChain(blocks) } + // Instrument the fetching and imported events + 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 } + // Announce the duplicating block, wait for retrieval, and also propagate directly tester.fetcher.Notify("valid", hashes[0], time.Now().Add(-arriveTimeout), fetcher) - time.Sleep(50 * time.Millisecond) + <-fetching tester.fetcher.Enqueue("valid", blocks[hashes[0]]) tester.fetcher.Enqueue("valid", blocks[hashes[0]]) @@ -358,8 +419,13 @@ func TestImportDeduplication(t *testing.T) { // Fill the missing block directly as if propagated, and check import uniqueness tester.fetcher.Enqueue("valid", blocks[hashes[1]]) - time.Sleep(50 * time.Millisecond) - + for done := false; !done; { + select { + case <-imported: + case <-time.After(50 * time.Millisecond): + done = true + } + } if imported := len(tester.blocks); imported != 3 { t.Fatalf("synchronised block mismatch: have %v, want %v", imported, 3) } @@ -400,8 +466,12 @@ func TestDistantDiscarding(t *testing.T) { // block announcements to a node, but that even in the face of such an attack, // the fetcher remains operational. func TestHashMemoryExhaustionAttack(t *testing.T) { + // Create a tester with instrumented import hooks tester := newTester() + imported := make(chan *types.Block) + tester.fetcher.importedHook = func(block *types.Block) { imported <- block } + // Create a valid chain and an infinite junk chain hashes := createHashes(hashLimit+2*maxQueueDist, knownHash) blocks := createBlocksFromHashes(hashes) @@ -413,29 +483,39 @@ func TestHashMemoryExhaustionAttack(t *testing.T) { // Feed the tester a huge hashset from the attacker, and a limited from the valid peer for i := 0; i < len(attack); i++ { if i < maxQueueDist { - tester.fetcher.Notify("valid", hashes[len(hashes)-1-i], time.Now().Add(arriveTimeout/2), valid) + tester.fetcher.Notify("valid", hashes[len(hashes)-2-i], time.Now(), valid) } - tester.fetcher.Notify("attacker", attack[i], time.Now().Add(arriveTimeout/2), attacker) + tester.fetcher.Notify("attacker", attack[i], time.Now(), attacker) } if len(tester.fetcher.announced) != hashLimit+maxQueueDist { t.Fatalf("queued announce count mismatch: have %d, want %d", len(tester.fetcher.announced), hashLimit+maxQueueDist) } - // Wait for synchronisation to complete and check success for the valid peer - time.Sleep(2 * arriveTimeout) - if imported := len(tester.blocks); imported != maxQueueDist { - t.Fatalf("partial synchronised block mismatch: have %v, want %v", imported, maxQueueDist) + // Wait for fetches to complete + for i := 0; i < maxQueueDist; i++ { + select { + case <-imported: + case <-time.After(time.Second): + t.Fatalf("block %d: import timeout", i) + } + } + select { + case <-imported: + t.Fatalf("extra block imported") + case <-time.After(50 * time.Millisecond): } // Feed the remaining valid hashes to ensure DOS protection state remains clean - for i := len(hashes) - maxQueueDist; i >= 0; { - for j := 0; j < maxQueueDist && i >= 0; j++ { - tester.fetcher.Notify("valid", hashes[i], time.Now().Add(time.Millisecond), valid) - i-- + for i := len(hashes) - maxQueueDist - 2; i >= 0; i-- { + tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), valid) + select { + case <-imported: + case <-time.After(time.Second): + t.Fatalf("block %d: import timeout", len(hashes)-i) } - time.Sleep(500 * time.Millisecond) } - time.Sleep(500 * time.Millisecond) - if imported := len(tester.blocks); imported != len(hashes) { - t.Fatalf("fully synchronised block mismatch: have %v, want %v", imported, len(hashes)) + select { + case <-imported: + t.Fatalf("extra block imported") + case <-time.After(50 * time.Millisecond): } } @@ -443,14 +523,18 @@ func TestHashMemoryExhaustionAttack(t *testing.T) { // announces and retrievals) don't pile up indefinitely, exhausting available // system memory. func TestBlockMemoryExhaustionAttack(t *testing.T) { + // Create a tester with instrumented import hooks tester := newTester() + imported := make(chan *types.Block) + tester.fetcher.importedHook = func(block *types.Block) { imported <- block } + // Create a valid chain and a batch of dangling (but in range) blocks - hashes := createHashes(blockLimit, knownHash) + hashes := createHashes(blockLimit+2*maxQueueDist, knownHash) blocks := createBlocksFromHashes(hashes) attack := make(map[common.Hash]*types.Block) - for i := 0; i < 16; i++ { + for len(attack) < blockLimit+2*maxQueueDist { hashes := createHashes(maxQueueDist-1, unknownHash) blocks := createBlocksFromHashes(hashes) for _, hash := range hashes[:maxQueueDist-2] { @@ -475,18 +559,27 @@ func TestBlockMemoryExhaustionAttack(t *testing.T) { } // Insert the missing piece (and sanity check the import) tester.fetcher.Enqueue("valid", blocks[hashes[len(hashes)-2]]) - time.Sleep(500 * time.Millisecond) - if imported := len(tester.blocks); imported != maxQueueDist+1 { - t.Fatalf("synchronised block mismatch: have %v, want %v", imported, maxQueueDist+1) + for i := 0; i < maxQueueDist; i++ { + select { + case <-imported: + case <-time.After(time.Second): + t.Fatalf("block %d: import timeout", i) + } + } + select { + case <-imported: + t.Fatalf("extra block imported") + case <-time.After(50 * time.Millisecond): } // Insert the remaining blocks in chunks to ensure clean DOS protection for i := maxQueueDist; i < len(hashes)-1; i++ { tester.fetcher.Enqueue("valid", blocks[hashes[len(hashes)-2-i]]) - if i%maxQueueDist == 0 { - time.Sleep(500 * time.Millisecond) + select { + case <-imported: + case <-time.After(time.Second): + t.Fatalf("block %d: import timeout", len(hashes)-i) } } - time.Sleep(500 * time.Millisecond) if imported := len(tester.blocks); imported != len(hashes) { t.Fatalf("synchronised block mismatch: have %v, want %v", imported, len(hashes)) } From 99ca4b619b22c000dfe633b036ed0b9b0fe83523 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Mon, 22 Jun 2015 18:28:38 +0300 Subject: [PATCH 4/5] eth/fetcher: clean up test assertions --- eth/fetcher/fetcher_test.go | 160 +++++++++++------------------------- 1 file changed, 49 insertions(+), 111 deletions(-) diff --git a/eth/fetcher/fetcher_test.go b/eth/fetcher/fetcher_test.go index 06ca2ef86f..80247d9d20 100644 --- a/eth/fetcher/fetcher_test.go +++ b/eth/fetcher/fetcher_test.go @@ -159,6 +159,37 @@ func (f *fetcherTester) makeFetcher(blocks map[common.Hash]*types.Block) blockRe } } +// verifyImportEvent verifies that one single event arrive on an import channel. +func verifyImportEvent(t *testing.T, imported chan *types.Block) { + select { + case <-imported: + case <-time.After(time.Second): + t.Fatalf("import timeout") + } +} + +// verifyImportCount verifies that exactly count number of events arrive on an +// import hook channel. +func verifyImportCount(t *testing.T, imported chan *types.Block, count int) { + for i := 0; i < count; i++ { + select { + case <-imported: + case <-time.After(time.Second): + t.Fatalf("block %d: import timeout", i) + } + } + verifyImportDone(t, imported) +} + +// verifyImportDone verifies that no more events are arriving on an import channel. +func verifyImportDone(t *testing.T, imported chan *types.Block) { + select { + case <-imported: + t.Fatalf("extra block imported") + case <-time.After(50 * time.Millisecond): + } +} + // Tests that a fetcher accepts block announcements and initiates retrievals for // them, successfully importing into the local chain. func TestSequentialAnnouncements(t *testing.T) { @@ -176,18 +207,9 @@ func TestSequentialAnnouncements(t *testing.T) { for i := len(hashes) - 2; i >= 0; i-- { tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), fetcher) - - select { - case <-imported: - case <-time.After(time.Second): - t.Fatalf("block %d: import timeout", len(hashes)-i) - } - } - select { - case <-imported: - t.Fatalf("extra block imported") - case <-time.After(50 * time.Millisecond): + verifyImportEvent(t, imported) } + verifyImportDone(t, imported) } // Tests that if blocks are announced by multiple peers (or even the same buggy @@ -216,17 +238,10 @@ func TestConcurrentAnnouncements(t *testing.T) { tester.fetcher.Notify("second", hashes[i], time.Now().Add(-arriveTimeout+time.Millisecond), wrapper) tester.fetcher.Notify("second", hashes[i], time.Now().Add(-arriveTimeout-time.Millisecond), wrapper) - select { - case <-imported: - case <-time.After(time.Second): - t.Fatalf("block %d: import timeout", len(hashes)-i) - } - } - select { - case <-imported: - t.Fatalf("extra block imported") - case <-time.After(50 * time.Millisecond): + verifyImportEvent(t, imported) } + verifyImportDone(t, imported) + // Make sure no blocks were retrieved twice if int(counter) != targetBlocks { t.Fatalf("retrieval count mismatch: have %v, want %v", counter, targetBlocks) @@ -259,18 +274,7 @@ func TestOverlappingAnnouncements(t *testing.T) { } } // Wait for all the imports to complete and check count - for i := 0; i < len(hashes)-1; i++ { - select { - case <-imported: - case <-time.After(time.Second): - t.Fatalf("block %d: import timeout", i) - } - } - select { - case <-imported: - t.Fatalf("extra block imported") - case <-time.After(50 * time.Millisecond): - } + verifyImportCount(t, imported, len(hashes)-1) } // Tests that announces already being retrieved will not be duplicated. @@ -334,19 +338,7 @@ func TestRandomArrivalImport(t *testing.T) { } // Finally announce the skipped entry and check full import tester.fetcher.Notify("valid", hashes[skip], time.Now().Add(-arriveTimeout), fetcher) - - for i := 0; i < len(hashes)-1; i++ { - select { - case <-imported: - case <-time.After(time.Second): - t.Fatalf("block %d: import timeout", i) - } - } - select { - case <-imported: - t.Fatalf("extra block imported") - case <-time.After(50 * time.Millisecond): - } + verifyImportCount(t, imported, len(hashes)-1) } // Tests that direct block enqueues (due to block propagation vs. hash announce) @@ -372,19 +364,7 @@ func TestQueueGapFill(t *testing.T) { } // Fill the missing block directly as if propagated tester.fetcher.Enqueue("valid", blocks[hashes[skip]]) - - for i := 0; i < len(hashes)-1; i++ { - select { - case <-imported: - case <-time.After(time.Second): - t.Fatalf("block %d: import timeout", i) - } - } - select { - case <-imported: - t.Fatalf("extra block imported") - case <-time.After(50 * time.Millisecond): - } + verifyImportCount(t, imported, len(hashes)-1) } // Tests that blocks arriving from various sources (multiple propagations, hash @@ -419,16 +399,8 @@ func TestImportDeduplication(t *testing.T) { // Fill the missing block directly as if propagated, and check import uniqueness tester.fetcher.Enqueue("valid", blocks[hashes[1]]) - for done := false; !done; { - select { - case <-imported: - case <-time.After(50 * time.Millisecond): - done = true - } - } - if imported := len(tester.blocks); imported != 3 { - t.Fatalf("synchronised block mismatch: have %v, want %v", imported, 3) - } + verifyImportCount(t, imported, 2) + if counter != 2 { t.Fatalf("import invocation count mismatch: have %v, want %v", counter, 2) } @@ -491,32 +463,14 @@ func TestHashMemoryExhaustionAttack(t *testing.T) { t.Fatalf("queued announce count mismatch: have %d, want %d", len(tester.fetcher.announced), hashLimit+maxQueueDist) } // Wait for fetches to complete - for i := 0; i < maxQueueDist; i++ { - select { - case <-imported: - case <-time.After(time.Second): - t.Fatalf("block %d: import timeout", i) - } - } - select { - case <-imported: - t.Fatalf("extra block imported") - case <-time.After(50 * time.Millisecond): - } + verifyImportCount(t, imported, maxQueueDist) + // Feed the remaining valid hashes to ensure DOS protection state remains clean for i := len(hashes) - maxQueueDist - 2; i >= 0; i-- { tester.fetcher.Notify("valid", hashes[i], time.Now().Add(-arriveTimeout), valid) - select { - case <-imported: - case <-time.After(time.Second): - t.Fatalf("block %d: import timeout", len(hashes)-i) - } - } - select { - case <-imported: - t.Fatalf("extra block imported") - case <-time.After(50 * time.Millisecond): + verifyImportEvent(t, imported) } + verifyImportDone(t, imported) } // Tests that blocks sent to the fetcher (either through propagation or via hash @@ -559,28 +513,12 @@ func TestBlockMemoryExhaustionAttack(t *testing.T) { } // Insert the missing piece (and sanity check the import) tester.fetcher.Enqueue("valid", blocks[hashes[len(hashes)-2]]) - for i := 0; i < maxQueueDist; i++ { - select { - case <-imported: - case <-time.After(time.Second): - t.Fatalf("block %d: import timeout", i) - } - } - select { - case <-imported: - t.Fatalf("extra block imported") - case <-time.After(50 * time.Millisecond): - } + verifyImportCount(t, imported, maxQueueDist) + // Insert the remaining blocks in chunks to ensure clean DOS protection for i := maxQueueDist; i < len(hashes)-1; i++ { tester.fetcher.Enqueue("valid", blocks[hashes[len(hashes)-2-i]]) - select { - case <-imported: - case <-time.After(time.Second): - t.Fatalf("block %d: import timeout", len(hashes)-i) - } - } - if imported := len(tester.blocks); imported != len(hashes) { - t.Fatalf("synchronised block mismatch: have %v, want %v", imported, len(hashes)) + verifyImportEvent(t, imported) } + verifyImportDone(t, imported) } From 3ce17d2862e89e20b8e9b10cc02ee0b1333f6625 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Mon, 22 Jun 2015 20:13:18 +0300 Subject: [PATCH 5/5] eth/fetcher: fix a closure data race --- eth/fetcher/fetcher.go | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go index 7b5804ab22..90a202235e 100644 --- a/eth/fetcher/fetcher.go +++ b/eth/fetcher/fetcher.go @@ -281,12 +281,13 @@ func (f *Fetcher) loop() { glog.V(logger.Detail).Infof("Peer %s: fetching %s", peer, list) } - hashes := hashes // closure! + // Create a closure of the fetch and schedule in on a new thread + fetcher, hashes := f.fetching[hashes[0]].fetch, hashes go func() { if f.fetchingHook != nil { f.fetchingHook(hashes) } - f.fetching[hashes[0]].fetch(hashes) + fetcher(hashes) }() } // Schedule the next fetch if blocks are still pending