eth/downloader: accumulating hash bans for reconnecting attackers

2 files changed, 126 insertions, 5 deletions

diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go
index de6ee1734..b0d55bc44 100644
--- a/eth/downloader/downloader.go
+++ b/eth/downloader/downloader.go
@@ -1,7 +1,9 @@
 package downloader
 
 import (
+	"bytes"
 	"errors"
+	"fmt"
 	"math/rand"
 	"sync"
 	"sync/atomic"
@@ -289,9 +291,15 @@ func (d *Downloader) fetchHashes(p *peer, h common.Hash) error {
 				glog.V(logger.Debug).Infof("Peer (%s) responded with empty hash set", active.id)
 				return ErrEmptyHashSet
 			}
-			for _, hash := range hashPack.hashes {
+			for index, hash := range hashPack.hashes {
 				if d.banned.Has(hash) {
 					glog.V(logger.Debug).Infof("Peer (%s) sent a known invalid chain", active.id)
+
+					d.queue.Insert(hashPack.hashes[:index+1])
+					if err := d.banBlocks(active.id, hash); err != nil {
+						fmt.Println("ban err", err)
+						glog.V(logger.Debug).Infof("Failed to ban batch of blocks: %v", err)
+					}
 					return ErrInvalidChain
 				}
 			}
@@ -399,8 +407,10 @@ func (d *Downloader) fetchBlocks() error {
 	glog.V(logger.Debug).Infoln("Downloading", d.queue.Pending(), "block(s)")
 	start := time.Now()
 
-	// default ticker for re-fetching blocks every now and then
+	// Start a ticker to continue throttled downloads and check for bad peers
 	ticker := time.NewTicker(20 * time.Millisecond)
+	defer ticker.Stop()
+
 out:
 	for {
 		select {
@@ -413,7 +423,7 @@ out:
 				block := blockPack.blocks[0]
 				if _, ok := d.checks[block.Hash()]; ok {
 					delete(d.checks, block.Hash())
-					continue
+					break
 				}
 			}
 			// If the peer was previously banned and failed to deliver it's pack
@@ -488,7 +498,7 @@ out:
 			if d.queue.Pending() > 0 {
 				// Throttle the download if block cache is full and waiting processing
 				if d.queue.Throttle() {
-					continue
+					break
 				}
 				// Send a download request to all idle peers, until throttled
 				idlePeers := d.peers.IdlePeers()
@@ -529,10 +539,86 @@ out:
 		}
 	}
 	glog.V(logger.Detail).Infoln("Downloaded block(s) in", time.Since(start))
-
 	return nil
 }
 
+// banBlocks retrieves a batch of blocks from a peer feeding us invalid hashes,
+// and bans the head of the retrieved batch.
+//
+// This method only fetches one single batch as the goal is not ban an entire
+// (potentially long) invalid chain - wasting a lot of time in the meanwhile -,
+// but rather to gradually build up a blacklist if the peer keeps reconnecting.
+func (d *Downloader) banBlocks(peerId string, head common.Hash) error {
+	glog.V(logger.Debug).Infof("Banning a batch out of %d blocks from %s", d.queue.Pending(), peerId)
+
+	// Ask the peer being banned for a batch of blocks from the banning point
+	peer := d.peers.Peer(peerId)
+	if peer == nil {
+		return nil
+	}
+	request := d.queue.Reserve(peer, MaxBlockFetch)
+	if request == nil {
+		return nil
+	}
+	if err := peer.Fetch(request); err != nil {
+		return err
+	}
+	// Wait a bit for the reply to arrive, and ban if done so
+	timeout := time.After(blockTTL)
+	for {
+		select {
+		case <-d.cancelCh:
+			return errCancelBlockFetch
+
+		case <-timeout:
+			return ErrTimeout
+
+		case blockPack := <-d.blockCh:
+			blocks := blockPack.blocks
+
+			// Short circuit if it's a stale cross check
+			if len(blocks) == 1 {
+				block := blocks[0]
+				if _, ok := d.checks[block.Hash()]; ok {
+					delete(d.checks, block.Hash())
+					break
+				}
+			}
+			// Short circuit if it's not from the peer being banned
+			if blockPack.peerId != peerId {
+				break
+			}
+			// Short circuit if no blocks were returned
+			if len(blocks) == 0 {
+				return errors.New("no blocks returned to ban")
+			}
+			// Got the batch of invalid blocks, reconstruct their chain order
+			for i := 0; i < len(blocks); i++ {
+				for j := i + 1; j < len(blocks); j++ {
+					if blocks[i].NumberU64() > blocks[j].NumberU64() {
+						blocks[i], blocks[j] = blocks[j], blocks[i]
+					}
+				}
+			}
+			// Ensure we're really banning the correct blocks
+			if bytes.Compare(blocks[0].Hash().Bytes(), head.Bytes()) != 0 {
+				return errors.New("head block not the banned one")
+			}
+			index := 0
+			for _, block := range blocks[1:] {
+				if bytes.Compare(block.ParentHash().Bytes(), blocks[index].Hash().Bytes()) != 0 {
+					break
+				}
+				index++
+			}
+			d.banned.Add(blocks[index].Hash())
+
+			glog.V(logger.Debug).Infof("Banned %d blocks from: %s\n", index+1, peerId)
+			return nil
+		}
+	}
+}
+
 // DeliverBlocks injects a new batch of blocks received from a remote node.
 // This is usually invoked through the BlocksMsg by the protocol handler.
 func (d *Downloader) DeliverBlocks(id string, blocks []*types.Block) error {
diff --git a/eth/downloader/downloader_test.go b/eth/downloader/downloader_test.go
index 434861c61..288072164 100644
--- a/eth/downloader/downloader_test.go
+++ b/eth/downloader/downloader_test.go
@@ -14,6 +14,7 @@ import (
 var (
 	knownHash   = common.Hash{1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}
 	unknownHash = common.Hash{9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9}
+	bannedHash  = common.Hash{5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5}
 )
 
 func createHashes(start, amount int) (hashes []common.Hash) {
@@ -520,3 +521,37 @@ func TestMadeupParentBlockChainAttack(t *testing.T) {
 		t.Fatalf("failed to synchronise blocks: %v", err)
 	}
 }
+
+// Tests that if one/multiple malicious peers try to feed a banned blockchain to
+// the downloader, it will not keep refetching the same chain indefinitely, but
+// gradually block pieces of it, until it's head is also blocked.
+func TestBannedChainStarvationAttack(t *testing.T) {
+	// Construct a valid chain, but ban one of the hashes in it
+	hashes := createHashes(0, 8*blockCacheLimit)
+	hashes[len(hashes)/2+23] = bannedHash // weird index to have non multiple of ban chunk size
+
+	blocks := createBlocksFromHashes(hashes)
+
+	// Create the tester and ban the selected hash
+	tester := newTester(t, hashes, blocks)
+	tester.downloader.banned.Add(bannedHash)
+
+	// Iteratively try to sync, and verify that the banned hash list grows until
+	// the head of the invalid chain is blocked too.
+	tester.newPeer("attack", big.NewInt(10000), hashes[0])
+	for banned := tester.downloader.banned.Size(); ; {
+		// Try to sync with the attacker, check hash chain failure
+		if _, err := tester.syncTake("attack", hashes[0]); err != ErrInvalidChain {
+			t.Fatalf("synchronisation error mismatch: have %v, want %v", err, ErrInvalidChain)
+		}
+		// Check that the ban list grew with at least 1 new item, or all banned
+		bans := tester.downloader.banned.Size()
+		if bans < banned+1 {
+			if tester.downloader.banned.Has(hashes[0]) {
+				break
+			}
+			t.Fatalf("ban count mismatch: have %v, want %v+", bans, banned+1)
+		}
+		banned = bans
+	}
+}