ETCM-541 attempt UPnP port mapping to aid in peer discovery & connection (#929)

Author: jvdp

nix/pkgs/mantis.nix

@@ -50,7 +50,7 @@ in sbt.mkDerivation {
 
   # This sha represents the change dependencies of mantis.
   # Update this sha whenever you change the dependencies
-  depsSha256 = "1058ryh7nj7y59iwk60ap0kgky4j0awpfvq76p9l4picz9qgg9i8";
+  depsSha256 = "14hx1gxa7505b8jy1vq5gc5p51fn80sj0pafx26awsrl6q67qyld";
 
   # this is the command used to to create the fixed-output-derivation
   depsWarmupCommand = "sbt compile --debug -Dnix=true";

project/Dependencies.scala

@@ -114,7 +114,11 @@ object Dependencies {
     "org.scala-sbt.ipcsocket" % "ipcsocket" % "1.1.0",
     "org.xerial.snappy" % "snappy-java" % "1.1.7.7",
     "org.web3j" % "core" % "5.0.0" % Test,
-    "io.vavr" % "vavr" % "1.0.0-alpha-3"
+    "io.vavr" % "vavr" % "1.0.0-alpha-3",
+    "org.jupnp" % "org.jupnp" % "2.5.2",
+    "org.jupnp" % "org.jupnp.support" % "2.5.2",
+    "org.jupnp" % "org.jupnp.tool" % "2.5.2",
+    "javax.servlet" % "javax.servlet-api" % "4.0.1"
   )
 
   val guava: Seq[ModuleID] = {

scalastyle-config.xml

@@ -1,7 +1,7 @@
 <scalastyle>
     <name>Scalastyle standard configuration</name>
     <check level="error" class="org.scalastyle.file.FileTabChecker" enabled="true"></check>
-    <check level="error" class="org.scalastyle.file.FileLengthChecker" enabled="true">
+    <check level="error" class="org.scalastyle.file.FileLengthChecker" enabled="false">
         <parameters>
             <parameter name="maxFileLength"><![CDATA[800]]></parameter>
         </parameters>

src/main/resources/application.conf

@@ -47,6 +47,9 @@ mantis {
       port = 9076
     }
 
+    # Try automatic port forwarding via UPnP
+    automatic-port-forwarding = true
+
     discovery {
 
       # Turn discovery of/off
@@ -577,7 +580,7 @@ mantis {
       #           otherwise mantis VM will be run in the same process, but acting as an external VM (listening at `host` and `port`)
       # - none: doesn't run anything, expect the VM to be started by other means
       vm-type = "mantis"
-
+ 
       # path to the executable - optional depending on the `vm-type` setting
       executable-path = "./bin/mantis-vm"
 

src/main/resources/logback.xml

@@ -63,5 +63,7 @@
     <logger name="io.iohk.ethereum.network.PeerActor" level="${LOGSLEVEL}" />
     <logger name="io.iohk.ethereum.network.rlpx.RLPxConnectionHandler" level="${LOGSLEVEL}" />
     <logger name="io.iohk.ethereum.vm.VM" level="OFF" />
-
+    <logger name="org.jupnp.QueueingThreadPoolExecutor" level="WARN" />
+    <logger name="org.jupnp.util.SpecificationViolationReporter" level="ERROR" />
+    <logger name="org.jupnp.protocol.RetrieveRemoteDescriptors" level="ERROR" />
 </configuration>

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

@@ -0,0 +1,81 @@
+package io.iohk.ethereum.network
+
+import io.iohk.ethereum.utils.Logger
+import java.net.InetAddress
+import java.util.concurrent.ExecutorService
+import monix.eval.Task
+import org.jupnp.DefaultUpnpServiceConfiguration
+import org.jupnp.support.igd.PortMappingListener
+import org.jupnp.support.model.PortMapping
+import org.jupnp.support.model.PortMapping.Protocol.{TCP, UDP}
+import org.jupnp.tool.transport.JDKTransportConfiguration
+import org.jupnp.transport.Router
+import org.jupnp.transport.spi.NetworkAddressFactory
+import org.jupnp.transport.spi.StreamClient
+import org.jupnp.transport.spi.StreamClientConfiguration
+import org.jupnp.transport.spi.StreamServer
+import org.jupnp.transport.spi.StreamServerConfiguration
+import org.jupnp.UpnpServiceImpl
+import scala.jdk.CollectionConverters._
+import scala.util.chaining._
+import org.jupnp.QueueingThreadPoolExecutor
+import cats.effect.Resource
+import org.jupnp.UpnpService
+import cats.implicits._
+
+private class ClientOnlyUpnpServiceConfiguration extends DefaultUpnpServiceConfiguration() {
+  private final val THREAD_POOL_SIZE = 4 // seemingly the minimum required to perform port mapping
+
+  override def createDefaultExecutorService(): ExecutorService =
+    QueueingThreadPoolExecutor.createInstance("mantis-jupnp", THREAD_POOL_SIZE);
+
+  override def createStreamClient(): StreamClient[_ <: StreamClientConfiguration] =
+    JDKTransportConfiguration.INSTANCE.createStreamClient(getSyncProtocolExecutorService())
+
+  override def createStreamServer(networkAddressFactory: NetworkAddressFactory): NoStreamServer.type =
+    NoStreamServer // prevent a StreamServer from running needlessly
+}
+
+private object NoStreamServer extends StreamServer[StreamServerConfiguration] {
+  def run(): Unit = ()
+  def init(_1: InetAddress, _2: Router): Unit = ()
+  def getPort(): Int = 0
+  def stop(): Unit = ()
+  def getConfiguration(): StreamServerConfiguration = new StreamServerConfiguration {
+    def getListenPort(): Int = 0
+  }
+}
+
+object PortForwarder extends Logger {
+  private final val description = "Mantis"
+
+  def openPorts(tcpPorts: Seq[Int], udpPorts: Seq[Int]): Resource[Task, Unit] =
+    Resource.make(startForwarding(tcpPorts, udpPorts))(stopForwarding).void
+
+  private def startForwarding(tcpPorts: Seq[Int], udpPorts: Seq[Int]): Task[UpnpService] = Task {
+    log.info("Attempting port forwarding for TCP ports {} and UDP ports {}", tcpPorts, udpPorts)
+    new UpnpServiceImpl(new ClientOnlyUpnpServiceConfiguration()).tap { service =>
+      service.startup()
+
+      val bindAddresses =
+        service
+          .getConfiguration()
+          .createNetworkAddressFactory()
+          .getBindAddresses()
+          .asScala
+          .map(_.getHostAddress())
+          .toArray
+
+      val portMappings = for {
+        address <- bindAddresses
+        (port, protocol) <- tcpPorts.map(_ -> TCP) ++ udpPorts.map(_ -> UDP)
+      } yield new PortMapping(port, address, protocol).tap(_.setDescription(description))
+
+      service.getRegistry().addListener(new PortMappingListener(portMappings))
+    }
+  }
+
+  private def stopForwarding(service: UpnpService) = Task {
+    service.shutdown()
+  }
+}

src/main/scala/io/iohk/ethereum/nodebuilder/NodeBuilder.scala

@@ -38,10 +38,13 @@ import java.util.concurrent.atomic.AtomicReference
 import io.iohk.ethereum.consensus.blocks.CheckpointBlockGenerator
 import org.bouncycastle.crypto.AsymmetricCipherKeyPair
 
+import scala.concurrent.Future
 import scala.concurrent.duration._
 import scala.util.{Failure, Success, Try}
 import akka.util.ByteString
 import monix.execution.Scheduler
+import cats.implicits._
+import monix.eval.Task
 
 // scalastyle:off number.of.types
 trait BlockchainConfigBuilder {
@@ -114,7 +117,7 @@ trait PeerDiscoveryManagerBuilder {
     with DiscoveryServiceBuilder
     with StorageBuilder =>
 
-  import monix.execution.Scheduler.Implicits.global
+  import Scheduler.Implicits.global
 
   lazy val peerDiscoveryManager: ActorRef = system.actorOf(
     PeerDiscoveryManager.props(
@@ -676,6 +679,33 @@ trait SyncControllerBuilder {
 
 }
 
+trait PortForwardingBuilder {
+  self: DiscoveryConfigBuilder =>
+
+  import Scheduler.Implicits.global
+
+  private val portForwarding = PortForwarder
+    .openPorts(
+      Seq(Config.Network.Server.port),
+      Seq(discoveryConfig.port).filter(_ => discoveryConfig.discoveryEnabled)
+    )
+    .whenA(Config.Network.automaticPortForwarding)
+    .allocated
+    .map(_._2)
+
+  // reference to a task that produces the release task,
+  // memoized to prevent running multiple port forwarders at once
+  private val portForwardingRelease = new AtomicReference(Option.empty[Task[Task[Unit]]])
+
+  def startPortForwarding(): Future[Unit] = {
+    portForwardingRelease.compareAndSet(None, Some(portForwarding.memoize))
+    portForwardingRelease.get().fold(Future.unit)(_.runToFuture.void)
+  }
+
+  def stopPortForwarding(): Future[Unit] =
+    portForwardingRelease.getAndSet(None).fold(Future.unit)(_.flatten.runToFuture)
+}
+
 trait ShutdownHookBuilder {
   self: Logger =>
   def shutdown(): Unit = {
@@ -778,3 +808,4 @@ trait Node
     with AsyncConfigBuilder
     with CheckpointBlockGeneratorBuilder
     with TransactionHistoryServiceBuilder.Default
+    with PortForwardingBuilder

src/main/scala/io/iohk/ethereum/nodebuilder/StdNode.scala

@@ -62,6 +62,7 @@ abstract class BaseNode extends Node {
 
     startPeerManager()
 
+    startPortForwarding()
     startServer()
 
     startSyncController()
@@ -93,6 +94,7 @@ abstract class BaseNode extends Node {
         shutdownTimeoutDuration
       )
     )
+    tryAndLogFailure(() => Await.ready(stopPortForwarding(), shutdownTimeoutDuration))
     if (jsonRpcConfig.ipcServerConfig.enabled) {
       tryAndLogFailure(() => jsonRpcIpcServer.close())
     }

src/main/scala/io/iohk/ethereum/utils/Config.scala

@@ -42,6 +42,8 @@ object Config {
 
     val protocolVersion = networkConfig.getInt("protocol-version")
 
+    val automaticPortForwarding = networkConfig.getBoolean("automatic-port-forwarding")
+
     object Server {
       private val serverConfig = networkConfig.getConfig("server-address")