Sdk 142 - Dns client

src/main/resources/reference.conf

@@ -105,6 +105,9 @@ scorex {
     # List of IP addresses of well known nodes.
     knownPeers = []
 
+    # List of known DNS seeder.
+    knownSeeders = []
+
     # Interval between GetPeers messages to be send by our node to a random one
     getPeersInterval = 2m
 

src/main/scala/scorex/core/app/Application.scala

@@ -1,12 +1,12 @@
 package scorex.core.app
 
-import java.net.InetSocketAddress
-
 import akka.actor.{ActorRef, ActorSystem}
 import akka.http.scaladsl.Http
 import akka.http.scaladsl.server.{ExceptionHandler, RejectionHandler, Route}
 import scorex.core.api.http.{ApiErrorHandler, ApiRejectionHandler, ApiRoute, CompositeHttpService}
 import scorex.core.network._
+import scorex.core.network.dns.DnsClientRef
+import scorex.core.network.dns.model.DnsClientInput
 import scorex.core.network.message._
 import scorex.core.network.peer.PeerManagerRef
 import scorex.core.settings.ScorexSettings
@@ -15,6 +15,7 @@ import scorex.core.utils.NetworkTimeProvider
 import scorex.core.{NodeViewHolder, PersistentNodeViewModifier}
 import scorex.util.ScorexLogging
 
+import java.net.InetSocketAddress
 import scala.concurrent.ExecutionContext
 
 trait Application extends ScorexLogging {
@@ -75,21 +76,24 @@ trait Application extends ScorexLogging {
     }
   }
 
-  val scorexContext = ScorexContext(
+  val scorexContext: ScorexContext = ScorexContext(
     messageSpecs = basicSpecs ++ additionalMessageSpecs,
     features = features,
     upnpGateway = upnpGateway,
     timeProvider = timeProvider,
     externalNodeAddress = externalSocketAddress
   )
 
-  val peerManagerRef = PeerManagerRef(settings, scorexContext)
+  val peerManagerRef: ActorRef = PeerManagerRef(settings, scorexContext)
 
   val networkControllerRef: ActorRef = NetworkControllerRef(
     "networkController", settings.network, peerManagerRef, scorexContext)
 
+  val dnsClientParams = new DnsClientInput(settings.network.knownSeeders)
+  val dnsClient: ActorRef = DnsClientRef(dnsClientParams)
+
   val peerSynchronizer: ActorRef = PeerSynchronizerRef("PeerSynchronizer",
-    networkControllerRef, peerManagerRef, settings.network, featureSerializers)
+    networkControllerRef, peerManagerRef, dnsClient, settings.network, featureSerializers)
 
   lazy val combinedRoute: Route = CompositeHttpService(actorSystem, apiRoutes, settings.restApi, swaggerConfig).compositeRoute
 

src/main/scala/scorex/core/network/NetworkController.scala

@@ -1,23 +1,23 @@
 package scorex.core.network
 
-import java.net._
-
 import akka.actor._
 import akka.io.Tcp._
 import akka.io.{IO, Tcp}
 import akka.pattern.ask
 import akka.util.Timeout
 import scorex.core.app.{ScorexContext, Version}
 import scorex.core.network.NodeViewSynchronizer.ReceivableMessages.{DisconnectedPeer, HandshakedPeer}
+import scorex.core.network.PeerSynchronizer.ReceivableMessages.GetNewPeers
 import scorex.core.network.message.Message.MessageCode
 import scorex.core.network.message.{Message, MessageSpec}
 import scorex.core.network.peer.PeerManager.ReceivableMessages._
-import scorex.core.network.peer.{LocalAddressPeerFeature, PeerInfo, PeerManager, PeersStatus, PenaltyType, SessionIdPeerFeature}
+import scorex.core.network.peer._
 import scorex.core.settings.NetworkSettings
 import scorex.core.utils.TimeProvider.Time
 import scorex.core.utils.{NetworkUtils, TimeProvider}
 import scorex.util.ScorexLogging
 
+import java.net._
 import scala.concurrent.ExecutionContext
 import scala.concurrent.duration._
 import scala.language.{existentials, postfixOps}
@@ -139,6 +139,9 @@ class NetworkController(settings: NetworkSettings,
 
     case Blacklisted(peerAddress) =>
       closeConnection(peerAddress)
+
+    case EmptyPeerDatabase() =>
+      context.system.eventStream.publish(GetNewPeers())
   }
 
   private def connectionEvents: Receive = {
@@ -501,6 +504,8 @@ object NetworkController {
 
     case object GetConnectedPeers
 
+    case class EmptyPeerDatabase()
+
     /**
       * Get p2p network status
       */

src/main/scala/scorex/core/network/PeerSynchronizer.scala

@@ -4,13 +4,17 @@ import akka.actor.{Actor, ActorRef, ActorSystem, Props}
 import akka.pattern.ask
 import akka.util.Timeout
 import scorex.core.network.NetworkController.ReceivableMessages.{PenalizePeer, RegisterMessageSpecs, SendToNetwork}
+import scorex.core.network.PeerSynchronizer.ReceivableMessages.{GetNewPeers, LookupResponse}
+import scorex.core.network.dns.DnsClient.ReceivableMessages.LookupRequest
+import scorex.core.network.dns.strategy.Strategy.LeastNodeQuantity
 import scorex.core.network.message.{GetPeersSpec, Message, MessageSpec, PeersSpec}
+import scorex.core.network.peer.PeerManager.ReceivableMessages.{AddOrUpdatePeer, AddPeerIfEmpty, SeenPeers}
 import scorex.core.network.peer.{PeerInfo, PenaltyType}
-import scorex.core.network.peer.PeerManager.ReceivableMessages.{AddPeerIfEmpty, SeenPeers}
 import scorex.core.settings.NetworkSettings
 import scorex.util.ScorexLogging
 import shapeless.syntax.typeable._
 
+import java.net.{InetAddress, InetSocketAddress}
 import scala.concurrent.ExecutionContext
 import scala.concurrent.duration._
 
@@ -19,6 +23,7 @@ import scala.concurrent.duration._
   */
 class PeerSynchronizer(val networkControllerRef: ActorRef,
                        peerManager: ActorRef,
+                       dnsClientRef: ActorRef,
                        settings: NetworkSettings,
                        featureSerializers: PeerFeature.Serializers)
                       (implicit ec: ExecutionContext) extends Actor with Synchronizer with ScorexLogging {
@@ -38,16 +43,35 @@ class PeerSynchronizer(val networkControllerRef: ActorRef,
 
     networkControllerRef ! RegisterMessageSpecs(Seq(GetPeersSpec, peersSpec), self)
 
+    context.system.eventStream.subscribe(self, classOf[GetNewPeers])
+
     val msg = Message[Unit](GetPeersSpec, Right(Unit), None)
     val stn = SendToNetwork(msg, SendToRandom)
     context.system.scheduler.scheduleWithFixedDelay(2.seconds, settings.getPeersInterval, networkControllerRef, stn)
+
+    performFirstDnsSeedersLookup()
+  }
+
+  private def performFirstDnsSeedersLookup(): Unit = {
+    self ! GetNewPeers()
   }
 
   override def receive: Receive = {
 
     // data received from a remote peer
     case Message(spec, Left(msgBytes), Some(source)) => parseAndHandle(spec, msgBytes, source)
 
+    // TODO: it could be a good idea to pass the peers left in the db or some other criteria
+    // TODO: to decide the lookup strategy
+    case GetNewPeers() =>
+      dnsClientRef ! LookupRequest(LeastNodeQuantity())
+
+    case LookupResponse(ipv4Addresses, _) =>
+      val defaultPort = settings.bindAddress.getPort
+      ipv4Addresses.foreach(
+        address => peerManager ! AddOrUpdatePeer(PeerInfo.fromAddress(new InetSocketAddress(address, defaultPort)))
+      )
+
     // fall-through method for reporting unhandled messages
     case nonsense: Any => log.warn(s"PeerSynchronizer: got unexpected input $nonsense from ${sender()}")
   }
@@ -82,16 +106,24 @@ class PeerSynchronizer(val networkControllerRef: ActorRef,
   }
 }
 
+object PeerSynchronizer {
+  object ReceivableMessages {
+    case class GetNewPeers()
+
+    case class LookupResponse(ipv4Addresses: Seq[InetAddress], ipv6Addresses: Seq[InetAddress])
+  }
+}
+
 object PeerSynchronizerRef {
-  def props(networkControllerRef: ActorRef, peerManager: ActorRef, settings: NetworkSettings,
+  def props(networkControllerRef: ActorRef, peerManager: ActorRef, dnsClientRef: ActorRef, settings: NetworkSettings,
             featureSerializers: PeerFeature.Serializers)(implicit ec: ExecutionContext): Props =
-    Props(new PeerSynchronizer(networkControllerRef, peerManager, settings, featureSerializers))
+    Props(new PeerSynchronizer(networkControllerRef, peerManager, dnsClientRef, settings, featureSerializers))
 
-  def apply(networkControllerRef: ActorRef, peerManager: ActorRef, settings: NetworkSettings,
+  def apply(networkControllerRef: ActorRef, peerManager: ActorRef, dnsClientRef: ActorRef, settings: NetworkSettings,
             featureSerializers: PeerFeature.Serializers)(implicit system: ActorSystem, ec: ExecutionContext): ActorRef =
-    system.actorOf(props(networkControllerRef, peerManager, settings, featureSerializers))
+    system.actorOf(props(networkControllerRef, peerManager, dnsClientRef, settings, featureSerializers))
 
-  def apply(name: String, networkControllerRef: ActorRef, peerManager: ActorRef, settings: NetworkSettings,
+  def apply(name: String, networkControllerRef: ActorRef, peerManager: ActorRef, dnsClientRef: ActorRef, settings: NetworkSettings,
             featureSerializers: PeerFeature.Serializers)(implicit system: ActorSystem, ec: ExecutionContext): ActorRef =
-    system.actorOf(props(networkControllerRef, peerManager, settings, featureSerializers), name)
+    system.actorOf(props(networkControllerRef, peerManager, dnsClientRef, settings, featureSerializers), name)
 }

src/main/scala/scorex/core/network/dns/DnsClient.scala

@@ -0,0 +1,50 @@
+package scorex.core.network.dns
+
+import akka.actor.{Actor, ActorRef, ActorSystem, Props}
+import scorex.core.network.dns.model.DnsClientInput
+import scorex.core.network.dns.strategy.LookupStrategy
+import scorex.util.ScorexLogging
+
+import scala.util.{Failure, Success}
+
+class DnsClient(dnsClientParams: DnsClientInput) extends Actor with ScorexLogging {
+
+  import scorex.core.network.dns.DnsClient.ReceivableMessages.LookupRequest
+
+  override def receive: Receive =
+    dnsLookup orElse nonsense
+
+  private def dnsLookup: Receive = {
+    case LookupRequest(strategy: LookupStrategy) =>
+      strategy.apply(dnsClientParams) match {
+        case Success(response) => sender() ! response
+        case Failure(exception) => log.error("Failed to perform a dns lookup: " + exception.getMessage)
+      }
+  }
+
+  private def nonsense: Receive = {
+    case nonsense: Any =>
+      log.warn(s"DnsClient: got unexpected input $nonsense")
+  }
+}
+
+object DnsClient {
+  object ReceivableMessages {
+    case class LookupRequest(s: LookupStrategy)
+  }
+
+  object Exception {
+    final case class DnsLookupException(private val message: String = "")
+      extends Exception(message)
+  }
+}
+
+object DnsClientRef {
+  def props(params: DnsClientInput): Props = {
+    Props(new DnsClient(params))
+  }
+
+  def apply(dnsClientParams: DnsClientInput)(implicit system: ActorSystem): ActorRef = {
+    system.actorOf(props(dnsClientParams))
+  }
+}

src/main/scala/scorex/core/network/dns/model/DnsClientInput.scala

@@ -0,0 +1,14 @@
+package scorex.core.network.dns.model
+
+import scorex.core.network.dns.model.DnsClientInput.defaultLookupFunction
+
+import java.net.InetAddress
+import scala.util.Try
+
+case class DnsClientInput(dnsSeeders: Seq[DnsSeederDomain], lookupFunction: DnsSeederDomain => Try[Seq[InetAddress]] = defaultLookupFunction)
+
+object DnsClientInput {
+  def defaultLookupFunction(url: DnsSeederDomain): Try[Seq[InetAddress]] = {
+    Try(InetAddress.getAllByName(url.domainName))
+  }
+}

src/main/scala/scorex/core/network/dns/model/DnsSeederDomain.scala

@@ -0,0 +1,10 @@
+package scorex.core.network.dns.model
+
+class DnsSeederDomain(val domainName: String) {
+  private val WEB_DOMAIN_REGEX = "^(((?!\\-))(xn\\-\\-)?[a-z0-9\\-_]{0,61}[a-z0-9]{1,1}\\.)*(xn\\-\\-)?([a-z0-9\\-]{1,61}|[a-z0-9\\-]{1,30})\\.[a-z]{2,}$".r
+
+  domainName match {
+    case e if e.trim.isEmpty || WEB_DOMAIN_REGEX.findFirstMatchIn(e).isEmpty => throw new IllegalArgumentException("Domain name is not well formed")
+    case _ =>
+  }
+}

src/main/scala/scorex/core/network/dns/strategy/LookupStrategy.scala

@@ -0,0 +1,67 @@
+package scorex.core.network.dns.strategy
+
+import scorex.core.network.PeerSynchronizer.ReceivableMessages.LookupResponse
+import scorex.core.network.dns.DnsClient.Exception.DnsLookupException
+import scorex.core.network.dns.model.DnsClientInput
+
+import java.net.{Inet4Address, Inet6Address, InetAddress}
+import scala.util.{Failure, Success, Try}
+
+trait LookupStrategy {
+  def apply(dnsClientParams: DnsClientInput): Try[LookupResponse]
+}
+
+object Strategy {
+  case class LeastNodeQuantity() extends LookupStrategy {
+    private val LEAST_NODE_QUANTITY = 1
+
+    override def apply(dnsClientParams: DnsClientInput): Try[LookupResponse] = thresholdNodeQuantity(dnsClientParams, LEAST_NODE_QUANTITY)
+  }
+
+  case class MaxNodeQuantity() extends LookupStrategy {
+    private val MAX_NODE_QUANTITY = Int.MaxValue
+
+    override def apply(dnsClientParams: DnsClientInput): Try[LookupResponse] = {
+      thresholdNodeQuantity(dnsClientParams, MAX_NODE_QUANTITY)
+    }
+  }
+
+  case class ThresholdNodeQuantity(nodeThreshold: Int) extends LookupStrategy {
+    override def apply(dnsClientParams: DnsClientInput): Try[LookupResponse] = thresholdNodeQuantity(dnsClientParams, nodeThreshold)
+  }
+
+  private def thresholdNodeQuantity(dnsClientParams: DnsClientInput, nodesThreshold: Int): Try[LookupResponse] = {
+    if (nodesThreshold <= 0) {
+      Failure(new IllegalArgumentException("The nodes threshold must be greater than 0"))
+    } else {
+      val seeders = dnsClientParams.dnsSeeders
+      val lookupFunction = dnsClientParams.lookupFunction
+
+      val exceptionMessages = Seq[String]()
+
+      val (ipv4Addresses, ipv6Addresses) = seeders.foldLeft(Seq[InetAddress]()) { (acc, curr) =>
+        val newElements = if (acc.size < nodesThreshold) {
+          lookupFunction(curr) match {
+            case Success(value) => value
+            case Failure(exception) =>
+              exceptionMessages ++ exception.getMessage
+              Seq[InetAddress]()
+          }
+        } else {
+          Seq[InetAddress]()
+        }
+
+        acc ++ newElements
+      }.partition {
+        case _: Inet4Address => true
+        case _: Inet6Address => false
+      }
+
+      if (ipv4Addresses.isEmpty && ipv6Addresses.isEmpty) {
+        Failure(DnsLookupException("All dns lookups failed: " + exceptionMessages.mkString(", ")))
+      } else {
+        Success(LookupResponse(ipv4Addresses, ipv6Addresses))
+      }
+    }
+  }
+}

src/main/scala/scorex/core/network/peer/PeerManager.scala

@@ -1,9 +1,9 @@
 package scorex.core.network.peer
 
 import java.net.{InetAddress, InetSocketAddress}
-
 import akka.actor.{Actor, ActorRef, ActorSystem, Props}
 import scorex.core.app.ScorexContext
+import scorex.core.network.NetworkController.ReceivableMessages.EmptyPeerDatabase
 import scorex.core.network._
 import scorex.core.settings.ScorexSettings
 import scorex.core.utils.NetworkUtils
@@ -23,7 +23,11 @@ class PeerManager(settings: ScorexSettings, scorexContext: ScorexContext) extend
 
   if (peerDatabase.isEmpty) {
     // fill database with peers from config file if empty
-    settings.network.knownPeers.foreach { address =>
+    fillPeerDatabase(settings.network.knownPeers)
+  }
+
+  private def fillPeerDatabase(addresses: Seq[InetSocketAddress]): Unit = {
+    addresses.foreach { address =>
       if (!isSelf(address)) {
         peerDatabase.addOrUpdateKnownPeer(PeerInfo.fromAddress(address))
       }
@@ -65,6 +69,9 @@ class PeerManager(settings: ScorexSettings, scorexContext: ScorexContext) extend
     case RemovePeer(address) =>
       log.info(s"$address removed from peers database")
       peerDatabase.remove(address)
+      if (peerDatabaseHasOnlyKnownPeersFromConfig()) {
+        sender() ! EmptyPeerDatabase
+      }
 
     case get: GetPeers[_] =>
       sender() ! get.choose(peerDatabase.knownPeers, peerDatabase.blacklistedPeers, scorexContext)
@@ -91,6 +98,8 @@ class PeerManager(settings: ScorexSettings, scorexContext: ScorexContext) extend
     peerSpec.declaredAddress.exists(isSelf) || peerSpec.localAddressOpt.exists(isSelf)
   }
 
+  private def peerDatabaseHasOnlyKnownPeersFromConfig(): Boolean = peerDatabase.knownPeers.size <= settings.network.knownPeers.size
+
 }
 
 object PeerManager {