[ETCM-355] Peer fork id validation

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

@@ -14,10 +14,10 @@ import io.iohk.ethereum.network.Peer
 import io.iohk.ethereum.network.PeerId
 import io.iohk.ethereum.network.p2p.MessageSerializable
 import io.iohk.ethereum.network.p2p.messages.BaseETH6XMessages
+import io.iohk.ethereum.network.p2p.messages.Capability
 import io.iohk.ethereum.network.p2p.messages.ETC64
 import io.iohk.ethereum.network.p2p.messages.ETH62
 import io.iohk.ethereum.network.p2p.messages.ETH62.BlockHash
-import io.iohk.ethereum.network.p2p.messages.ProtocolFamily._
 
 class BlockBroadcast(val etcPeerManager: ActorRef) {
 
@@ -47,9 +47,10 @@ class BlockBroadcast(val etcPeerManager: ActorRef) {
     obtainRandomPeerSubset(peers.values.map(_.peer).toSet).foreach { peer =>
       val remoteStatus = peers(peer.id).peerInfo.remoteStatus
 
-      val message: MessageSerializable = remoteStatus.protocolFamily match {
-        case ETH => blockToBroadcast.as63
-        case ETC => blockToBroadcast.as64
+      val message: MessageSerializable = remoteStatus.capability match {
+        case Capability.ETH63 => blockToBroadcast.as63
+        case Capability.ETH64 => blockToBroadcast.as63
+        case Capability.ETC64 => blockToBroadcast.asEtc64
       }
       etcPeerManager ! EtcPeerManagerActor.SendMessage(message, peer.id)
     }
@@ -80,6 +81,6 @@ object BlockBroadcast {
     */
   case class BlockToBroadcast(block: Block, chainWeight: ChainWeight) {
     def as63: BaseETH6XMessages.NewBlock = BaseETH6XMessages.NewBlock(block, chainWeight.totalDifficulty)
-    def as64: ETC64.NewBlock = ETC64.NewBlock(block, chainWeight)
+    def asEtc64: ETC64.NewBlock = ETC64.NewBlock(block, chainWeight)
   }
 }

src/main/scala/io/iohk/ethereum/forkid/ForkIdValidator.scala

@@ -22,9 +22,12 @@ case object Connect extends ForkIdValidationResult
 case object ErrRemoteStale extends ForkIdValidationResult
 case object ErrLocalIncompatibleOrStale extends ForkIdValidationResult
 
+import cats.effect._
+
 object ForkIdValidator {
 
-  implicit val unsafeLogger: SelfAwareStructuredLogger[Task] = Slf4jLogger.getLogger[Task]
+  implicit val taskLogger: SelfAwareStructuredLogger[Task] = Slf4jLogger.getLogger[Task]
+  implicit val syncIoLogger: SelfAwareStructuredLogger[SyncIO] = Slf4jLogger.getLogger[SyncIO]
 
   val maxUInt64: BigInt = (BigInt(0x7fffffffffffffffL) << 1) + 1 // scalastyle:ignore magic.number
 

src/main/scala/io/iohk/ethereum/network/EtcPeerManagerActor.scala

@@ -20,16 +20,14 @@ import io.iohk.ethereum.network.handshaker.Handshaker.HandshakeResult
 import io.iohk.ethereum.network.p2p.Message
 import io.iohk.ethereum.network.p2p.MessageSerializable
 import io.iohk.ethereum.network.p2p.messages.BaseETH6XMessages
+import io.iohk.ethereum.network.p2p.messages.Capability
 import io.iohk.ethereum.network.p2p.messages.Codes
 import io.iohk.ethereum.network.p2p.messages.ETC64
 import io.iohk.ethereum.network.p2p.messages.ETC64.NewBlock
 import io.iohk.ethereum.network.p2p.messages.ETH62.BlockHeaders
 import io.iohk.ethereum.network.p2p.messages.ETH62.GetBlockHeaders
 import io.iohk.ethereum.network.p2p.messages.ETH62.NewBlockHashes
 import io.iohk.ethereum.network.p2p.messages.ETH64
-import io.iohk.ethereum.network.p2p.messages.ProtocolFamily
-import io.iohk.ethereum.network.p2p.messages.ProtocolFamily.ETC
-import io.iohk.ethereum.network.p2p.messages.ProtocolFamily.ETH
 import io.iohk.ethereum.network.p2p.messages.WireProtocol.Disconnect
 import io.iohk.ethereum.utils.ByteStringUtils
 
@@ -243,17 +241,15 @@ object EtcPeerManagerActor {
     * (they are different versions of Status msg)
     */
   case class RemoteStatus(
-      protocolFamily: ProtocolFamily,
-      protocolVersion: Int,
+      capability: Capability,
       networkId: Int,
       chainWeight: ChainWeight,
       bestHash: ByteString,
       genesisHash: ByteString
   ) {
     override def toString: String =
       s"RemoteStatus { " +
-        s"protocolFamily: $protocolFamily, " +
-        s"protocolVersion: $protocolVersion, " +
+        s"capability: $capability, " +
         s"networkId: $networkId, " +
         s"chainWeight: $chainWeight, " +
         s"bestHash: ${ByteStringUtils.hash2string(bestHash)}, " +
@@ -264,8 +260,7 @@ object EtcPeerManagerActor {
   object RemoteStatus {
     def apply(status: ETH64.Status): RemoteStatus =
       RemoteStatus(
-        ETH,
-        status.protocolVersion,
+        Capability.ETH64,
         status.networkId,
         ChainWeight.totalDifficultyOnly(status.totalDifficulty),
         status.bestHash,
@@ -274,8 +269,7 @@ object EtcPeerManagerActor {
 
     def apply(status: ETC64.Status): RemoteStatus =
       RemoteStatus(
-        ETC,
-        status.protocolVersion,
+        Capability.ETC64,
         status.networkId,
         status.chainWeight,
         status.bestHash,
@@ -284,8 +278,7 @@ object EtcPeerManagerActor {
 
     def apply(status: BaseETH6XMessages.Status): RemoteStatus =
       RemoteStatus(
-        ETH,
-        status.protocolVersion,
+        Capability.ETH63,
         status.networkId,
         ChainWeight.totalDifficultyOnly(status.totalDifficulty),
         status.bestHash,

src/main/scala/io/iohk/ethereum/network/handshaker/EtcHelloExchangeState.scala

@@ -6,7 +6,6 @@ import io.iohk.ethereum.network.EtcPeerManagerActor.PeerInfo
 import io.iohk.ethereum.network.handshaker.Handshaker.NextMessage
 import io.iohk.ethereum.network.p2p.Message
 import io.iohk.ethereum.network.p2p.messages.Capability
-import io.iohk.ethereum.network.p2p.messages.ProtocolVersions
 import io.iohk.ethereum.network.p2p.messages.WireProtocol.Disconnect
 import io.iohk.ethereum.network.p2p.messages.WireProtocol.Hello
 import io.iohk.ethereum.utils.Config
@@ -31,21 +30,21 @@ case class EtcHelloExchangeState(handshakerConfiguration: EtcHandshakerConfigura
     log.debug("Protocol handshake finished with peer ({})", hello)
     // FIXME in principle this should be already negotiated
     Capability.negotiate(hello.capabilities.toList, handshakerConfiguration.blockchainConfig.capabilities) match {
-      case Some(ProtocolVersions.ETC64) =>
+      case Some(Capability.ETC64) =>
         log.debug("Negotiated protocol version with client {} is etc/64", hello.clientId)
         EtcNodeStatus64ExchangeState(handshakerConfiguration)
-      case Some(ProtocolVersions.ETH63) =>
+      case Some(Capability.ETH63) =>
         log.debug("Negotiated protocol version with client {} is eth/63", hello.clientId)
         EthNodeStatus63ExchangeState(handshakerConfiguration)
-      case Some(ProtocolVersions.ETH64) =>
+      case Some(Capability.ETH64) =>
         log.debug("Negotiated protocol version with client {} is eth/64", hello.clientId)
         EthNodeStatus64ExchangeState(handshakerConfiguration)
       case _ =>
         log.debug(
           s"Connected peer does not support {} / {} / {} protocol. Disconnecting.",
-          ProtocolVersions.ETH63,
-          ProtocolVersions.ETH64,
-          ProtocolVersions.ETC64
+          Capability.ETH63,
+          Capability.ETH64,
+          Capability.ETC64
         )
         DisconnectedState(Disconnect.Reasons.IncompatibleP2pProtocolVersion)
     }

src/main/scala/io/iohk/ethereum/network/handshaker/EtcNodeStatus64ExchangeState.scala

@@ -4,8 +4,8 @@ import io.iohk.ethereum.network.EtcPeerManagerActor.PeerInfo
 import io.iohk.ethereum.network.EtcPeerManagerActor.RemoteStatus
 import io.iohk.ethereum.network.p2p.Message
 import io.iohk.ethereum.network.p2p.MessageSerializable
+import io.iohk.ethereum.network.p2p.messages.Capability
 import io.iohk.ethereum.network.p2p.messages.ETC64
-import io.iohk.ethereum.network.p2p.messages.ProtocolVersions
 
 case class EtcNodeStatus64ExchangeState(
     handshakerConfiguration: EtcHandshakerConfiguration
@@ -22,7 +22,7 @@ case class EtcNodeStatus64ExchangeState(
     val chainWeight = blockchain.getChainWeightByHash(bestBlockHeader.hash).get
 
     val status = ETC64.Status(
-      protocolVersion = ProtocolVersions.ETC64.version,
+      protocolVersion = Capability.ETC64.version,
       networkId = peerConfiguration.networkId,
       chainWeight = chainWeight,
       bestHash = bestBlockHeader.hash,

src/main/scala/io/iohk/ethereum/network/handshaker/EthNodeStatus63ExchangeState.scala

@@ -5,7 +5,7 @@ import io.iohk.ethereum.network.EtcPeerManagerActor.RemoteStatus
 import io.iohk.ethereum.network.p2p.Message
 import io.iohk.ethereum.network.p2p.MessageSerializable
 import io.iohk.ethereum.network.p2p.messages.BaseETH6XMessages
-import io.iohk.ethereum.network.p2p.messages.ProtocolVersions
+import io.iohk.ethereum.network.p2p.messages.Capability
 
 case class EthNodeStatus63ExchangeState(
     handshakerConfiguration: EtcHandshakerConfiguration
@@ -23,7 +23,7 @@ case class EthNodeStatus63ExchangeState(
     val chainWeight = blockchain.getChainWeightByHash(bestBlockHeader.hash).get
 
     val status = BaseETH6XMessages.Status(
-      protocolVersion = ProtocolVersions.ETH63.version,
+      protocolVersion = Capability.ETH63.version,
       networkId = peerConfiguration.networkId,
       totalDifficulty = chainWeight.totalDifficulty,
       bestHash = bestBlockHeader.hash,

src/main/scala/io/iohk/ethereum/network/handshaker/EthNodeStatus64ExchangeState.scala

@@ -1,12 +1,17 @@
 package io.iohk.ethereum.network.handshaker
 
+import cats.effect._
+
+import io.iohk.ethereum.forkid.Connect
 import io.iohk.ethereum.forkid.ForkId
+import io.iohk.ethereum.forkid.ForkIdValidator
 import io.iohk.ethereum.network.EtcPeerManagerActor.PeerInfo
 import io.iohk.ethereum.network.EtcPeerManagerActor.RemoteStatus
 import io.iohk.ethereum.network.p2p.Message
 import io.iohk.ethereum.network.p2p.MessageSerializable
+import io.iohk.ethereum.network.p2p.messages.Capability
 import io.iohk.ethereum.network.p2p.messages.ETH64
-import io.iohk.ethereum.network.p2p.messages.ProtocolVersions
+import io.iohk.ethereum.network.p2p.messages.WireProtocol.Disconnect
 
 case class EthNodeStatus64ExchangeState(
     handshakerConfiguration: EtcHandshakerConfiguration
@@ -15,8 +20,18 @@ case class EthNodeStatus64ExchangeState(
   import handshakerConfiguration._
 
   def applyResponseMessage: PartialFunction[Message, HandshakerState[PeerInfo]] = { case status: ETH64.Status =>
-    // TODO: validate fork id of the remote peer
-    applyRemoteStatusMessage(RemoteStatus(status))
+    import ForkIdValidator.syncIoLogger
+    (for {
+      validationResult <-
+        ForkIdValidator.validatePeer[SyncIO](blockchainReader.genesisHeader.hash, blockchainConfig)(
+          blockchainReader.getBestBlockNumber(),
+          status.forkId
+        )
+      _ = log.info("Fork id validation result was: {}", validationResult)
+    } yield validationResult match {
+      case Connect => applyRemoteStatusMessage(RemoteStatus(status))
+      case _       => DisconnectedState[PeerInfo](Disconnect.Reasons.UselessPeer)
+    }).unsafeRunSync()
   }
 
   override protected def createStatusMsg(): MessageSerializable = {
@@ -25,7 +40,7 @@ case class EthNodeStatus64ExchangeState(
     val genesisHash = blockchainReader.genesisHeader.hash
 
     val status = ETH64.Status(
-      protocolVersion = ProtocolVersions.ETH64.version,
+      protocolVersion = Capability.ETH64.version,
       networkId = peerConfiguration.networkId,
       totalDifficulty = chainWeight.totalDifficulty,
       bestHash = bestBlockHeader.hash,

src/main/scala/io/iohk/ethereum/network/p2p/MessageDecoders.scala

@@ -109,9 +109,9 @@ object ETH63MessageDecoder extends MessageDecoder {
 object EthereumMessageDecoder {
   def ethMessageDecoder(protocolVersion: Capability): MessageDecoder =
     protocolVersion match {
-      case Capability.Capabilities.Etc64Capability => ETC64MessageDecoder.orElse(NetworkMessageDecoder)
-      case Capability.Capabilities.Eth63Capability => ETH63MessageDecoder.orElse(NetworkMessageDecoder)
-      case Capability.Capabilities.Eth64Capability => ETH64MessageDecoder.orElse(NetworkMessageDecoder)
-      case _                                       => throw new RuntimeException(s"Unsupported Protocol Version $protocolVersion")
+      case Capability.ETC64 => ETC64MessageDecoder.orElse(NetworkMessageDecoder)
+      case Capability.ETH63 => ETH63MessageDecoder.orElse(NetworkMessageDecoder)
+      case Capability.ETH64 => ETH64MessageDecoder.orElse(NetworkMessageDecoder)
+      case _                => throw new RuntimeException(s"Unsupported Protocol Version $protocolVersion")
     }
 }

src/main/scala/io/iohk/ethereum/network/p2p/messages/Capability.scala

@@ -6,54 +6,61 @@ import io.iohk.ethereum.rlp.RLPImplicitConversions._
 import io.iohk.ethereum.rlp.RLPImplicits._
 import io.iohk.ethereum.rlp.RLPList
 import io.iohk.ethereum.rlp.RLPSerializable
+import io.iohk.ethereum.rlp.RLPValue
 import io.iohk.ethereum.rlp.rawDecode
 
-case class Capability(name: String, version: Byte)
+sealed trait ProtocolFamily
+object ProtocolFamily {
+  final case object ETH extends ProtocolFamily
+  final case object ETC extends ProtocolFamily
+  implicit class ProtocolFamilyEnc(val msg: ProtocolFamily) extends RLPSerializable {
+    override def toRLPEncodable: RLPEncodeable = msg match {
+      case ETH => RLPValue("eth".getBytes())
+      case ETC => RLPValue("etc".getBytes())
+    }
+  }
+}
+
+sealed abstract class Capability(val name: ProtocolFamily, val version: Byte)
 
 object Capability {
+  case object ETH63 extends Capability(ProtocolFamily.ETH, 63) //scalastyle:ignore magic.number
+  case object ETH64 extends Capability(ProtocolFamily.ETH, 64) //scalastyle:ignore magic.number
+  case object ETC64 extends Capability(ProtocolFamily.ETC, 64) //scalastyle:ignore magic.number
+
+  def parse(s: String): Option[Capability] = s match {
+    case "eth/63" => Some(ETH63)
+    case "eth/64" => Some(ETH64)
+    case "etc/64" => Some(ETC64)
+    case _        => None // TODO: log unknown capability?
+  }
+
+  def parseUnsafe(s: String): Capability =
+    parse(s).getOrElse(throw new RuntimeException(s"Capability $s not supported by Mantis"))
+
   def negotiate(c1: List[Capability], c2: List[Capability]): Option[Capability] =
     c1.intersect(c2) match {
       case Nil => None
       case l   => Some(best(l))
     }
 
-  private val pattern = "(.*)/(\\d*)".r
-
-  def parseUnsafe(protocolVersion: String): Capability =
-    protocolVersion match {
-      case pattern(name, version) =>
-        val c = Capability(name, version.toByte)
-        if (Capabilities.All.contains(c))
-          c
-        else
-          throw new RuntimeException(s"Capability $protocolVersion not supported by Mantis")
-      case _ => throw new RuntimeException(s"Unable to parse capability $protocolVersion")
-    }
-
   //TODO consider how this scoring should be handled with 'snap' and other extended protocols
   def best(capabilities: List[Capability]): Capability =
     capabilities.maxBy(_.version)
 
   implicit class CapabilityEnc(val msg: Capability) extends RLPSerializable {
-    override def toRLPEncodable: RLPEncodeable = RLPList(msg.name, msg.version)
+    override def toRLPEncodable: RLPEncodeable = RLPList(msg.name.toRLPEncodable, msg.version)
   }
 
   implicit class CapabilityDec(val bytes: Array[Byte]) extends AnyVal {
-    def toCapability: Capability = CapabilityRLPEncodableDec(rawDecode(bytes)).toCapability
+    def toCapability: Option[Capability] = CapabilityRLPEncodableDec(rawDecode(bytes)).toCapability
   }
 
   implicit class CapabilityRLPEncodableDec(val rLPEncodeable: RLPEncodeable) extends AnyVal {
-    def toCapability: Capability = rLPEncodeable match {
-      case RLPList(name, version) => Capability(name, version)
+    def toCapability: Option[Capability] = rLPEncodeable match {
+      case RLPList(name, version) => parse(s"${stringEncDec.decode(name)}/${byteEncDec.decode(version)}")
       case _                      => throw new RLPException("Cannot decode Capability")
     }
   }
 
-  object Capabilities {
-    val Eth63Capability: Capability = ProtocolVersions.ETH63
-    val Eth64Capability: Capability = ProtocolVersions.ETH64
-    val Etc64Capability: Capability = ProtocolVersions.ETC64
-
-    val All: Seq[Capability] = Seq(ProtocolVersions.ETC64, ProtocolVersions.ETH63, ProtocolVersions.ETH64)
-  }
 }

src/main/scala/io/iohk/ethereum/network/p2p/messages/WireProtocol.scala

@@ -34,7 +34,7 @@ object WireProtocol {
 
       def toHello: Hello = rawDecode(bytes) match {
         case RLPList(p2pVersion, clientId, (capabilities: RLPList), listenPort, nodeId, _*) =>
-          Hello(p2pVersion, clientId, capabilities.items.map(_.toCapability), listenPort, nodeId)
+          Hello(p2pVersion, clientId, capabilities.items.map(_.toCapability).flatten, listenPort, nodeId)
         case _ => throw new RuntimeException("Cannot decode Hello")
       }
     }