[ETCM-331] Add Rate Limit for rpc requests

src/main/resources/application.conf

@@ -201,6 +201,19 @@ mantis {
         # Domains allowed to query RPC endpoint. Use "*" to enable requests from
         # any domain.
         cors-allowed-origins = []
+
+        # Rate Limit for JSON-RPC requests
+        # Limits the amount of request the same ip can perform in a given amount of time
+        rate-limit {
+          # If enabled, restrictions are applied
+          enabled = false
+
+          # Time that should pass between requests
+          min-request-interval = 10.seconds
+
+          # Size of stored timestamps for requests made from each ip
+          latest-timestamp-cache-size = 1024
+        }
       }
 
       ipc {

src/main/scala/io/iohk/ethereum/jsonrpc/server/http/BasicJsonRpcHttpServer.scala → src/main/scala/io/iohk/ethereum/jsonrpc/server/http/InsecureJsonRpcHttpServer.scala

@@ -11,10 +11,10 @@ import io.iohk.ethereum.utils.Logger
 import scala.concurrent.ExecutionContext.Implicits.global
 import scala.util.{Failure, Success}
 
-class BasicJsonRpcHttpServer(
+class InsecureJsonRpcHttpServer(
     val jsonRpcController: JsonRpcBaseController,
     val jsonRpcHealthChecker: JsonRpcHealthChecker,
-    config: JsonRpcHttpServerConfig
+    val config: JsonRpcHttpServerConfig
 )(implicit val actorSystem: ActorSystem)
     extends JsonRpcHttpServer
     with Logger {

src/main/scala/io/iohk/ethereum/jsonrpc/server/http/JsonRpcHttpServer.scala

@@ -11,19 +11,25 @@ import ch.megard.akka.http.cors.scaladsl.CorsDirectives._
 import ch.megard.akka.http.cors.scaladsl.model.HttpOriginMatcher
 import ch.megard.akka.http.cors.scaladsl.settings.CorsSettings
 import de.heikoseeberger.akkahttpjson4s.Json4sSupport
+import io.iohk.ethereum.faucet.jsonrpc.FaucetJsonRpcController
 import io.iohk.ethereum.jsonrpc._
 import io.iohk.ethereum.security.SSLError
 import io.iohk.ethereum.jsonrpc.serialization.JsonSerializers
 import io.iohk.ethereum.jsonrpc.server.controllers.JsonRpcBaseController
+import io.iohk.ethereum.jsonrpc.server.http.JsonRpcHttpServer.JsonRpcHttpServerConfig
 import io.iohk.ethereum.utils.{ConfigUtils, Logger}
 import javax.net.ssl.SSLContext
 import monix.eval.Task
 import monix.execution.Scheduler.Implicits.global
 import org.json4s.{DefaultFormats, JInt, native}
+import com.typesafe.config.{Config => TypesafeConfig}
 
-trait JsonRpcHttpServer extends Json4sSupport {
+import scala.concurrent.duration.{FiniteDuration, _}
+
+trait JsonRpcHttpServer extends Json4sSupport with RateLimit {
   val jsonRpcController: JsonRpcBaseController
   val jsonRpcHealthChecker: JsonRpcHealthChecker
+  val config: JsonRpcHttpServerConfig
 
   implicit val serialization = native.Serialization
 
@@ -50,14 +56,29 @@ trait JsonRpcHttpServer extends Json4sSupport {
     (path("healthcheck") & pathEndOrSingleSlash & get) {
       handleHealthcheck()
     } ~ (pathEndOrSingleSlash & post) {
-      entity(as[JsonRpcRequest]) { request =>
-        handleRequest(request)
+      (extractClientIP & entity(as[JsonRpcRequest])) { (clientAddress, request) =>
+        handleRequest(clientAddress, request)
       } ~ entity(as[Seq[JsonRpcRequest]]) { request =>
         handleBatchRequest(request)
       }
     }
   }
 
+  def handleRequest(clientAddress: RemoteAddress, request: JsonRpcRequest): StandardRoute = {
+    //FIXME: FaucetJsonRpcController.Status should be part of a Healthcheck request or alike.
+    // As a temporary solution, it is being excluded from the Rate Limit.
+    if (config.rateLimit.enabled && request.method != FaucetJsonRpcController.Status) {
+      handleRateLimitedRequest(clientAddress, request)
+    } else complete(jsonRpcController.handleRequest(request).runToFuture)
+  }
+
+  def handleRateLimitedRequest(clientAddress: RemoteAddress, request: JsonRpcRequest): StandardRoute = {
+    if (isRequestAvailable(clientAddress)) {
+      log.warn(s"Request limit exceeded for ip ${clientAddress.toIP.getOrElse("unknown")}")
+      complete(jsonRpcController.handleRequest(request).runToFuture)
+    } else complete(StatusCodes.TooManyRequests)
+  }
+
   /**
     * Try to start JSON RPC server
     */
@@ -86,11 +107,13 @@ trait JsonRpcHttpServer extends Json4sSupport {
   }
 
   private def handleBatchRequest(requests: Seq[JsonRpcRequest]) = {
-    complete {
-      Task
-        .traverse(requests)(request => jsonRpcController.handleRequest(request))
-        .runToFuture
-    }
+    if (!config.rateLimit.enabled) {
+      complete {
+        Task
+          .traverse(requests)(request => jsonRpcController.handleRequest(request))
+          .runToFuture
+      }
+    } else complete(StatusCodes.MethodNotAllowed)
   }
 }
 
@@ -104,27 +127,42 @@ object JsonRpcHttpServer extends Logger {
       fSslContext: () => Either[SSLError, SSLContext]
   )(implicit actorSystem: ActorSystem): Either[String, JsonRpcHttpServer] =
     config.mode match {
-      case "http" => Right(new BasicJsonRpcHttpServer(jsonRpcController, jsonRpcHealthchecker, config)(actorSystem))
+      case "http" => Right(new InsecureJsonRpcHttpServer(jsonRpcController, jsonRpcHealthchecker, config)(actorSystem))
       case "https" =>
         Right(
-          new JsonRpcHttpsServer(jsonRpcController, jsonRpcHealthchecker, config, secureRandom, fSslContext)(
+          new SecureJsonRpcHttpServer(jsonRpcController, jsonRpcHealthchecker, config, secureRandom, fSslContext)(
             actorSystem
           )
         )
       case _ => Left(s"Cannot start JSON RPC server: Invalid mode ${config.mode} selected")
     }
 
+  trait RateLimitConfig {
+    val enabled: Boolean
+    val minRequestInterval: FiniteDuration
+    val latestTimestampCacheSize: Int
+  }
+
+  object RateLimitConfig {
+    def apply(rateLimitConfig: TypesafeConfig): RateLimitConfig =
+      new RateLimitConfig {
+        override val enabled: Boolean = rateLimitConfig.getBoolean("enabled")
+        override val minRequestInterval: FiniteDuration =
+          rateLimitConfig.getDuration("min-request-interval").toMillis.millis
+        override val latestTimestampCacheSize: Int = rateLimitConfig.getInt("latest-timestamp-cache-size")
+      }
+  }
+
   trait JsonRpcHttpServerConfig {
     val mode: String
     val enabled: Boolean
     val interface: String
     val port: Int
     val corsAllowedOrigins: HttpOriginMatcher
+    val rateLimit: RateLimitConfig
   }
 
   object JsonRpcHttpServerConfig {
-    import com.typesafe.config.{Config => TypesafeConfig}
-
     def apply(mantisConfig: TypesafeConfig): JsonRpcHttpServerConfig = {
       val rpcHttpConfig = mantisConfig.getConfig("network.rpc.http")
 
@@ -135,6 +173,8 @@ object JsonRpcHttpServer extends Logger {
         override val port: Int = rpcHttpConfig.getInt("port")
 
         override val corsAllowedOrigins = ConfigUtils.parseCorsAllowedOrigins(rpcHttpConfig, "cors-allowed-origins")
+
+        override val rateLimit = RateLimitConfig(rpcHttpConfig.getConfig("rate-limit"))
       }
     }
   }

src/main/scala/io/iohk/ethereum/jsonrpc/server/http/RateLimit.scala

@@ -0,0 +1,26 @@
+package io.iohk.ethereum.jsonrpc.server.http
+
+import java.time.Clock
+
+import akka.http.scaladsl.model.RemoteAddress
+import com.twitter.util.LruMap
+import io.iohk.ethereum.jsonrpc.server.http.JsonRpcHttpServer.JsonRpcHttpServerConfig
+import io.iohk.ethereum.utils.Logger
+
+trait RateLimit extends Logger {
+
+  val config: JsonRpcHttpServerConfig
+
+  val latestRequestTimestamps = new LruMap[RemoteAddress, Long](config.rateLimit.latestTimestampCacheSize)
+
+  val clock: Clock = Clock.systemUTC()
+
+  def isRequestAvailable(clientAddress: RemoteAddress): Boolean = {
+    val timeMillis = clock.instant().toEpochMilli
+    val latestRequestTimestamp = latestRequestTimestamps.getOrElse(clientAddress, 0L)
+
+    val response = latestRequestTimestamp + config.rateLimit.minRequestInterval.toMillis < timeMillis
+    if (response) latestRequestTimestamps.put(clientAddress, timeMillis)
+    response
+  }
+}

src/main/scala/io/iohk/ethereum/jsonrpc/server/http/JsonRpcHttpsServer.scala → src/main/scala/io/iohk/ethereum/jsonrpc/server/http/SecureJsonRpcHttpServer.scala

@@ -1,24 +1,24 @@
 package io.iohk.ethereum.jsonrpc.server.http
 
-import java.security.SecureRandom
-
 import akka.actor.ActorSystem
 import akka.http.scaladsl.{ConnectionContext, Http}
 import ch.megard.akka.http.cors.scaladsl.model.HttpOriginMatcher
 import io.iohk.ethereum.jsonrpc.JsonRpcHealthChecker
 import io.iohk.ethereum.security.SSLError
-import io.iohk.ethereum.jsonrpc.server.controllers.JsonRpcBaseController
 import io.iohk.ethereum.jsonrpc.server.http.JsonRpcHttpServer.JsonRpcHttpServerConfig
+import java.security.SecureRandom
+
+import io.iohk.ethereum.jsonrpc.server.controllers.JsonRpcBaseController
 import io.iohk.ethereum.utils.Logger
 import javax.net.ssl.SSLContext
 
 import scala.concurrent.ExecutionContext.Implicits.global
 import scala.util.{Failure, Success}
 
-class JsonRpcHttpsServer(
+class SecureJsonRpcHttpServer(
     val jsonRpcController: JsonRpcBaseController,
     val jsonRpcHealthChecker: JsonRpcHealthChecker,
-    config: JsonRpcHttpServerConfig,
+    override val config: JsonRpcHttpServerConfig,
     secureRandom: SecureRandom,
     getSSLContext: () => Either[SSLError, SSLContext]
 )(implicit val actorSystem: ActorSystem)