eth/fetcher: handle and (crude) test block memory DOS

1 files changed, 45 insertions, 16 deletions

diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go
index 8d707da5c..ceca79df0 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)
+	}
+}