ETCM-370: Refine responses validation on block fetcher side and base blacklisting on that

Author: AnastasiiaL

src/main/scala/io/iohk/ethereum/blockchain/sync/Blacklist.scala

@@ -163,12 +163,16 @@ object Blacklist {
         val code: Int = 27
         val name: String = "UnrequestedBodies"
       }
-      case object RegularSyncRequestFailedType extends BlacklistReasonType with RegularSyncBlacklistGroup {
+      case object UnrequestedHeadersType extends BlacklistReasonType with RegularSyncBlacklistGroup {
         val code: Int = 28
+        val name: String = "UnrequestedHeaders"
+      }
+      case object RegularSyncRequestFailedType extends BlacklistReasonType with RegularSyncBlacklistGroup {
+        val code: Int = 29
         val name: String = "RegularSyncRequestFailed"
       }
       case object BlockImportErrorType extends BlacklistReasonType with RegularSyncBlacklistGroup {
-        val code: Int = 29
+        val code: Int = 30
         val name: String = "BlockImportError"
       }
     }
@@ -281,6 +285,10 @@ object Blacklist {
       val reasonType: BlacklistReasonType = UnrequestedBodiesType
       val description: String = "Received unrequested bodies"
     }
+    case object UnrequestedHeaders extends BlacklistReason {
+      val reasonType: BlacklistReasonType = UnrequestedHeadersType
+      val description: String = "Received unrequested headers"
+    }
     final case class RegularSyncRequestFailed(error: String) extends BlacklistReason {
       val reasonType: BlacklistReasonType = RegularSyncRequestFailedType
       val description: String = s"Request failed with error: $error"

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

@@ -9,10 +9,7 @@ import cats.instances.option._
 import io.iohk.ethereum.blockchain.sync.Blacklist.BlacklistReason
 import io.iohk.ethereum.consensus.validators.BlockValidator
 import io.iohk.ethereum.blockchain.sync.PeersClient._
-import io.iohk.ethereum.blockchain.sync.regular.BlockFetcherState.{
-  AwaitingBodiesToBeIgnored,
-  AwaitingHeadersToBeIgnored
-}
+import io.iohk.ethereum.blockchain.sync.regular.BlockFetcherState.{AwaitingBodiesToBeIgnored, AwaitingHeadersToBeIgnored, HeadersNotFormingSeq, HeadersNotMatchingReadyBlocks, HeadersNotMatchingWaitingHeaders}
 import io.iohk.ethereum.blockchain.sync.regular.BlockImporter.{ImportNewBlock, NotOnTop, OnTop}
 import io.iohk.ethereum.blockchain.sync.regular.RegularSync.ProgressProtocol
 import io.iohk.ethereum.domain._
@@ -130,7 +127,7 @@ class BlockFetcher(
         blockProvider.foreach(peersClient ! BlacklistPeer(_, BlacklistReason.BlockImportError(reason)))
         fetchBlocks(newState)
 
-      case ReceivedHeaders(headers) if state.isFetchingHeaders =>
+      case ReceivedHeaders(peer, headers) if state.isFetchingHeaders =>
         //First successful fetch
         if (state.waitingHeaders.isEmpty) {
           supervisor ! ProgressProtocol.StartedFetching
@@ -146,6 +143,14 @@ class BlockFetcher(
           } else {
             log.debug("Fetched {} headers starting from block {}", headers.size, headers.headOption.map(_.number))
             state.appendHeaders(headers) match {
+              case Left(HeadersNotFormingSeq.description) =>
+                log.info("Dismissed received headers due to: {}", HeadersNotFormingSeq.description)
+                peersClient ! BlacklistPeer(peer.id, BlacklistReason.UnrequestedHeaders)
+                state.withHeaderFetchReceived
+              case Left(HeadersNotMatchingReadyBlocks.description) =>
+                log.info("Dismissed received headers due to: {}", HeadersNotMatchingReadyBlocks.description)
+                peersClient ! BlacklistPeer(peer.id, BlacklistReason.UnrequestedHeaders)
+                state.withHeaderFetchReceived
               case Left(err) =>
                 log.info("Dismissed received headers due to: {}", err)
                 state.withHeaderFetchReceived
@@ -154,6 +159,10 @@ class BlockFetcher(
             }
           }
         fetchBlocks(newState)
+      case ReceivedHeaders(peer, _) if !state.isFetchingHeaders =>
+        peersClient ! BlacklistPeer(peer.id, BlacklistReason.UnrequestedHeaders)
+        Behaviors.same
+
       case RetryHeadersRequest if state.isFetchingHeaders =>
         log.debug("Something failed on a headers request, cancelling the request and re-fetching")
         fetchBlocks(state.withHeaderFetchReceived)
@@ -177,6 +186,10 @@ class BlockFetcher(
           fetchBlocks(newState)
         }
 
+      case ReceivedBodies(peer, _) if !state.isFetchingBodies =>
+        peersClient ! BlacklistPeer(peer.id, BlacklistReason.UnrequestedBodies)
+        Behaviors.same
+
       case RetryBodiesRequest if state.isFetchingBodies =>
         log.debug("Something failed on a bodies request, cancelling the request and re-fetching")
         fetchBlocks(state.withBodiesFetchReceived)
@@ -332,7 +345,7 @@ object BlockFetcher {
   final case object RetryBodiesRequest extends FetchCommand
   final case object RetryHeadersRequest extends FetchCommand
   final case class AdaptedMessageFromEventBus(message: Message, peerId: PeerId) extends FetchCommand
-  final case class ReceivedHeaders(headers: Seq[BlockHeader]) extends FetchCommand
+  final case class ReceivedHeaders(peer: Peer, headers: Seq[BlockHeader]) extends FetchCommand
   final case class ReceivedBodies(peer: Peer, bodies: Seq[BlockBody]) extends FetchCommand
 
   sealed trait FetchResponse

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

@@ -10,7 +10,6 @@ import io.iohk.ethereum.consensus.validators.BlockValidator
 import io.iohk.ethereum.domain.{Block, BlockBody, BlockHeader, HeadersSeq}
 import io.iohk.ethereum.network.PeerId
 import io.iohk.ethereum.network.p2p.messages.PV62.BlockHash
-import io.iohk.ethereum.utils.ByteStringUtils
 
 import scala.annotation.tailrec
 import scala.collection.immutable.Queue
@@ -86,7 +85,6 @@ case class BlockFetcherState(
 
   /**
     * Validates received headers consistency and their compatibility with the state
-    * TODO ETCM-370: This needs to be more fine-grained and detailed so blacklisting can be re-enabled
     */
   private def validatedHeaders(headers: Seq[BlockHeader]): Either[String, Seq[BlockHeader]] =
     if (headers.isEmpty) {
@@ -96,7 +94,7 @@ case class BlockFetcherState(
         .asRight[String]
         .ensure("Given headers should form a sequence without gaps")(HeadersSeq.areChain)
         .ensure("Given headers should form a sequence with ready blocks")(checkConsistencyWithReadyBlocks)
-        .ensure("Given headers do not form a chain with already stored ones")(headers =>
+        .ensure("Given headers should form a chain with waiting headers")(headers =>
           (waitingHeaders.lastOption, headers.headOption).mapN(_ isParentOf _).getOrElse(true)
         )
     }
@@ -319,4 +317,17 @@ object BlockFetcherState {
     * State used to keep track of pending request to prevent multiple requests in parallel
     */
   case object AwaitingBodiesToBeIgnored extends FetchingBodiesState
+
+  trait ValidationErrors {
+    def description: String
+  }
+  case object HeadersNotFormingSeq extends ValidationErrors {
+    val description = "Given headers should form a sequence without gaps"
+  }
+  case object HeadersNotMatchingReadyBlocks extends ValidationErrors {
+    val description = "Given headers should form a sequence with ready blocks"
+  }
+  case object HeadersNotMatchingWaitingHeaders extends ValidationErrors {
+    val description = "Given headers should form a chain with waiting headers"
+  }
 }

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

@@ -36,9 +36,9 @@ class HeadersFetcher(
         log.debug("Start fetching headers from block {}", blockNumber)
         requestHeaders(blockNumber, amount)
         Behaviors.same
-      case AdaptedMessage(_, BlockHeaders(headers)) =>
+      case AdaptedMessage(peer, BlockHeaders(headers)) =>
         log.debug("Fetched {} headers starting from block {}", headers.size, headers.headOption.map(_.number))
-        supervisor ! BlockFetcher.ReceivedHeaders(headers)
+        supervisor ! BlockFetcher.ReceivedHeaders(peer, headers)
         Behaviors.same
       case HeadersFetcher.RetryHeadersRequest =>
         supervisor ! BlockFetcher.RetryHeadersRequest

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

@@ -2,7 +2,6 @@ package io.iohk.ethereum.blockchain.sync.regular
 
 import akka.actor.{Actor, ActorLogging, ActorRef, AllForOneStrategy, Cancellable, Props, Scheduler, SupervisorStrategy}
 import akka.actor.typed.{ActorRef => TypedActorRef}
-import io.iohk.ethereum.blockchain.sync.SyncProtocol
 import io.iohk.ethereum.blockchain.sync.{Blacklist, SyncProtocol}
 import io.iohk.ethereum.blockchain.sync.SyncProtocol.Status
 import io.iohk.ethereum.blockchain.sync.SyncProtocol.Status.Progress

src/test/scala/io/iohk/ethereum/blockchain/sync/regular/RegularSyncSpec.scala

@@ -1,6 +1,7 @@
 package io.iohk.ethereum.blockchain.sync.regular
 
-import akka.actor.{ActorRef, ActorSystem}
+import akka.actor.{ActorRef, ActorSystem, PoisonPill}
+import akka.actor.typed.{ActorRef => TypedActorRef}
 import akka.testkit.TestActor.AutoPilot
 import akka.testkit.TestKit
 import akka.util.ByteString
@@ -10,6 +11,8 @@ import cats.syntax.traverse._
 import io.iohk.ethereum.blockchain.sync.Blacklist.BlacklistReason
 import io.iohk.ethereum.blockchain.sync.SyncProtocol.Status
 import io.iohk.ethereum.blockchain.sync.SyncProtocol.Status.Progress
+import io.iohk.ethereum.blockchain.sync.regular.BlockFetcher.{FetchCommand, PrintStatus, ReceivedHeaders}
+import io.iohk.ethereum.blockchain.sync.regular.HeadersFetcher.HeadersFetcherCommand
 import io.iohk.ethereum.blockchain.sync.regular.RegularSync.NewCheckpoint
 import io.iohk.ethereum.blockchain.sync.{PeersClient, SyncProtocol}
 import io.iohk.ethereum.crypto.kec256
@@ -37,6 +40,7 @@ import org.scalatest.{Assertion, BeforeAndAfterEach}
 
 import scala.concurrent.duration._
 import scala.concurrent.{Await, Future, Promise}
+import scala.math.BigInt
 
 class RegularSyncSpec
     extends WordSpecBase
@@ -112,34 +116,97 @@ class RegularSyncSpec
         peersClient.expectMsg(PeersClient.BlacklistPeer(defaultPeer.id, BlacklistReason.RegularSyncRequestFailed("a random reason")))
       })
 
-      //TODO: To be re-enabled with ETCM-370
-      "blacklist peer which returns headers starting from one with higher number than expected" ignore sync(
-        new Fixture(
-          testSystem
-        ) {
+      "blacklist peer which returns headers starting from one with higher number than expected" in sync(
+        new Fixture(testSystem) {
+          var blockFetcher: ActorRef = _
+
           regularSync ! SyncProtocol.Start
+          peerEventBus.expectMsgClass(classOf[Subscribe])
+          blockFetcher = peerEventBus.sender()
 
           peersClient.expectMsgEq(blockHeadersChunkRequest(0))
-          peersClient.reply(PeersClient.Response(defaultPeer, BlockHeaders(testBlocksChunked(1).headers)))
+          peersClient.reply(PeersClient.Response(defaultPeer, BlockHeaders(testBlocksChunked.head.headers)))
+
+          val getBodies: PeersClient.Request[GetBlockBodies] = PeersClient.Request.create(
+            GetBlockBodies(testBlocksChunked.head.headers.map(_.hash)),
+            PeersClient.BestPeer
+          )
+          peersClient.expectMsgEq(getBodies)
+          peersClient.reply(PeersClient.Response(defaultPeer, BlockBodies(testBlocksChunked.head.bodies)))
+
+          blockFetcher ! MessageFromPeer(NewBlock(testBlocks.last, ChainWeight.totalDifficultyOnly(testBlocks.last.number)), defaultPeer.id)
+          peersClient.expectMsgEq(blockHeadersChunkRequest(1))
+          peersClient.reply(PeersClient.Response(defaultPeer, BlockHeaders(testBlocksChunked(5).headers)))
           peersClient.expectMsgPF() {
             case PeersClient.BlacklistPeer(id, _) if id == defaultPeer.id => true
           }
         }
       )
 
-      //TODO: To be re-enabled with ETCM-370
-      "blacklist peer which returns headers not forming a chain" ignore sync(new Fixture(testSystem) {
+      "blacklist peer which returns headers not forming a chain" in sync(new Fixture(testSystem) {
         regularSync ! SyncProtocol.Start
 
         peersClient.expectMsgEq(blockHeadersChunkRequest(0))
         peersClient.reply(
-          PeersClient.Response(defaultPeer, BlockHeaders(testBlocksChunked.head.headers.filter(_.number % 2 == 0)))
+          PeersClient.Response(defaultPeer, BlockHeaders(testBlocks.headers.filter(_.number % 2 == 0)))
         )
         peersClient.expectMsgPF() {
           case PeersClient.BlacklistPeer(id, _) if id == defaultPeer.id => true
         }
       })
 
+      "blacklist peer which sends headers that were not requested" ignore sync(new Fixture(testSystem) {
+        import akka.actor.typed.scaladsl.adapter._
+        var blockFetcherAdapter: TypedActorRef[MessageFromPeer] = _
+        var blockFetcher: TypedActorRef[FetchCommand] = _
+//        var blockFetcher: ActorRef = _
+
+        regularSync ! SyncProtocol.Start
+        peerEventBus.expectMsgClass(classOf[Subscribe])
+        blockFetcherAdapter = peerEventBus.sender()
+
+        peersClient.expectMsgEq(blockHeadersChunkRequest(0))
+        peersClient.reply(PeersClient.Response(defaultPeer, BlockHeaders(testBlocksChunked.head.headers)))
+
+        val getBodies: PeersClient.Request[GetBlockBodies] = PeersClient.Request.create(
+          GetBlockBodies(testBlocksChunked.head.headers.map(_.hash)),
+          PeersClient.BestPeer
+        )
+        val getBodies2: PeersClient.Request[GetBlockBodies] = PeersClient.Request.create(
+          GetBlockBodies(testBlocksChunked(1).headers.map(_.hash)),
+          PeersClient.BestPeer
+        )
+        peersClient.expectMsgEq(getBodies)
+        peersClient.reply(PeersClient.Response(defaultPeer, BlockBodies(testBlocksChunked.head.bodies)))
+
+        blockFetcherAdapter ! MessageFromPeer(NewBlock(testBlocks(3), ChainWeight.totalDifficultyOnly(testBlocks(3).number)), defaultPeer.id)
+        peersClient.expectMsgEq(blockHeadersChunkRequest(1))
+        peersClient.reply(PeersClient.Response(defaultPeer, BlockHeaders(testBlocksChunked(2).headers)))
+
+        peersClient.expectMsgClass(classOf[PeersClient.BlacklistPeer])
+
+        blockFetcher = peersClient.sender()
+
+        println("============ "+ blockFetcher)
+        println("------------ "+ blockFetcherAdapter)
+        blockFetcher ! PrintStatus
+
+        val notBlacklistedPeer = peerByNumber(1)
+        peersClient.expectMsgEq(blockHeadersChunkRequest(1))
+        peersClient.reply(PeersClient.Response(notBlacklistedPeer, BlockHeaders(testBlocksChunked(1).headers)))
+        peersClient.expectMsgEq(getBodies2)
+        peersClient.reply(PeersClient.Response(notBlacklistedPeer, BlockBodies(testBlocksChunked(1).bodies)))
+
+        peersClient.expectNoMessage()
+
+        blockFetcher ! BlockFetcher.ReceivedHeaders(notBlacklistedPeer, testBlocksChunked(3).headers)
+        blockFetcher ! BlockFetcher.ReceivedBodies(notBlacklistedPeer, testBlocksChunked(3).bodies)
+
+        peersClient.expectMsgPF() {
+          case PeersClient.BlacklistPeer(id, _) if id == notBlacklistedPeer.id => true
+        }
+      })
+
       "wait for time defined in config until issuing a retry request due to no suitable peer" in sync(
         new Fixture(
           testSystem