diff options
| -rw-r--r-- | eth/downloader/downloader.go | 96 | ||||
| -rw-r--r-- | eth/downloader/downloader_test.go | 35 |
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 + } +} |