[ETCM-283] Add block bodies validation in block fetcher

Author: mirkoAlic

src/main/scala/io/iohk/ethereum/blockchain/sync/regular/BlockFetcher.scala

@@ -143,10 +143,15 @@ class BlockFetcher(
           log.debug("Received {} block bodies that will be ignored", bodies.size)
           state.withBodiesFetchReceived
         } else {
-          log.debug("Fetched {} block bodies", bodies.size)
-          state.withBodiesFetchReceived.addBodies(peer, bodies)
+          state.validateBodies(bodies) match {
+            case Left(err) =>
+              peersClient ! BlacklistPeer(peer.id, err)
+              state.withBodiesFetchReceived
+            case Right(newBlocks) =>
+              log.debug("Fetched {} block bodies", newBlocks.size)
+              state.withBodiesFetchReceived.appendNewBlocks(newBlocks, peer.id)
+          }
         }
-
       fetchBlocks(newState)
     case RetryBodiesRequest if state.isFetchingBodies =>
       log.debug("Time-out occurred while waiting for bodies")

src/main/scala/io/iohk/ethereum/blockchain/sync/regular/BlockFetcherState.scala

@@ -3,14 +3,17 @@ package io.iohk.ethereum.blockchain.sync.regular
 import akka.actor.ActorRef
 import akka.util.ByteString
 import cats.data.NonEmptyList
+//FIXME: By using this class, we are coupling sync process with a specific consensus (the standard one).
+import io.iohk.ethereum.consensus.validators.std.StdBlockValidator
 import io.iohk.ethereum.domain.{Block, BlockHeader, BlockBody, HeadersSeq}
-import io.iohk.ethereum.network.{Peer, PeerId}
+import io.iohk.ethereum.network.PeerId
 import io.iohk.ethereum.network.p2p.messages.PV62.BlockHash
 import BlockFetcherState._
 import cats.syntax.either._
 import cats.syntax.option._
 
 import scala.collection.immutable.Queue
+import scala.annotation.tailrec
 
 // scalastyle:off number.of.methods
 /**
@@ -101,14 +104,36 @@ case class BlockFetcherState(
         }
       )
 
-  def addBodies(peer: Peer, bodies: Seq[BlockBody]): BlockFetcherState = {
-    val (matching, waiting) = waitingHeaders.splitAt(bodies.length)
-    val blocks = matching.zip(bodies).map((Block.apply _).tupled)
+  def validateBodies(receivedBodies: Seq[BlockBody]): Either[String, Seq[Block]] =
+    bodiesAreOrderedSubsetOfRequested(waitingHeaders.toList, receivedBodies)
+      .toRight(
+        "Received unrequested bodies"
+      )
+
+  // Checks that the received block bodies are an ordered subset of the ones requested
+  @tailrec
+  private def bodiesAreOrderedSubsetOfRequested(
+      requestedHeaders: Seq[BlockHeader],
+      respondedBodies: Seq[BlockBody],
+      matchedBlocks: Seq[Block] = Nil
+  ): Option[Seq[Block]] =
+    (requestedHeaders, respondedBodies) match {
+      case (Seq(), _ +: _) => None
+      case (_, Seq()) => Some(matchedBlocks)
+      case (header +: remainingHeaders, body +: remainingBodies) =>
+        val doMatch = StdBlockValidator.validateHeaderAndBody(header, body).isRight
+        if (doMatch)
+          bodiesAreOrderedSubsetOfRequested(remainingHeaders, remainingBodies, matchedBlocks :+ Block(header, body))
+        else
+          bodiesAreOrderedSubsetOfRequested(remainingHeaders, respondedBodies, matchedBlocks)
+    }
 
-    withPeerForBlocks(peer.id, blocks.map(_.header.number))
+  def appendNewBlocks(blocks: Seq[Block], fromPeer: PeerId): BlockFetcherState = {
+    val receivedHeaders = blocks.map(_.header)
+    withPeerForBlocks(fromPeer, blocks.map(_.header.number))
       .copy(
-        readyBlocks = readyBlocks.enqueue(blocks),
-        waitingHeaders = waiting
+        readyBlocks = readyBlocks.enqueue(blocks.toList),
+        waitingHeaders = waitingHeaders.diff(receivedHeaders)
       )
   }