[JSON-API] Correctly extract the request source URL/IP (#10244)

* [JSON-API] Correctly extract the request source URL/IP changelog_begin - [JSON-API] If the service is put behind a proxy filling either of these headers X-Forwarded-For & X-Real-Ip then these will now be respected for logging the request source ip/url changelog_end * Return to the simple http server start code * Remove unused import * Update ledger-service/http-json/src/main/scala/com/digitalasset/http/Endpoints.scala Co-authored-by: Stephen Compall <[email protected]> Co-authored-by: Stephen Compall <[email protected]>
digital-asset · Jul 13, 2021 · b59f365 · b59f365
1 parent edaf541
commit b59f365
Show file tree

Hide file tree

Showing 4 changed files with 44 additions and 34 deletions.
diff --git a/ledger-service/http-json/BUILD.bazel b/ledger-service/http-json/BUILD.bazel
@@ -147,7 +147,10 @@ json_deps = {
 da_scala_binary(
     name = "http-json-binary",
     main_class = "com.daml.http.Main",
-    resources = [":src/main/resources/logback.xml"],
+    resources = [
+        ":src/main/resources/application.conf",
+        ":src/main/resources/logback.xml",
+    ],
     scala_deps = json_scala_deps,
     scalacopts = hj_scalacopts,
     tags = [
@@ -161,7 +164,10 @@ da_scala_binary(
 da_scala_binary(
     name = "http-json-binary-ee",
     main_class = "com.daml.http.Main",
-    resources = [":src/main/resources/logback.xml"],
+    resources = [
+        ":src/main/resources/application.conf",
+        ":src/main/resources/logback.xml",
+    ],
     scala_deps = json_scala_deps,
     scalacopts = hj_scalacopts,
     tags = [

diff --git a/ledger-service/http-json/src/main/resources/application.conf b/ledger-service/http-json/src/main/resources/application.conf
@@ -0,0 +1 @@
+akka.http.server.remote-address-attribute = on
diff --git a/ledger-service/http-json/src/main/scala/com/digitalasset/http/Endpoints.scala b/ledger-service/http-json/src/main/scala/com/digitalasset/http/Endpoints.scala
@@ -4,15 +4,16 @@
 package com.daml.http
 
 import akka.NotUsed
-import akka.http.scaladsl.Http
 import akka.http.scaladsl.model.HttpMethods.{GET, POST}
 import akka.http.scaladsl.model._
 import akka.http.scaladsl.model.headers.{
   Authorization,
   ModeledCustomHeader,
   ModeledCustomHeaderCompanion,
   OAuth2BearerToken,
+  `X-Forwarded-For`,
   `X-Forwarded-Proto`,
+  `X-Real-Ip`,
 }
 import akka.stream.Materializer
 import akka.stream.scaladsl.{Flow, Source}
@@ -67,12 +68,23 @@ class Endpoints(
   import util.ErrorOps._
   import Uri.Path._
 
+  // Inspired by
+  // https://github.com/akka/akka-http/blob/master/akka-http/src/main/scala/akka/http/scaladsl/server/directives/MiscDirectives.scala#L110-L116
+  // Because the Remote-Address header is deprecated we don't match for it here.
+  def requestSource(req: HttpRequest): RemoteAddress =
+    req
+      .header[`X-Forwarded-For`]
+      .flatMap(_.addresses.headOption)
+      .orElse(req.header[`X-Real-Ip`].map(_.address))
+      .orElse(req.attribute(AttributeKeys.remoteAddress))
+      .getOrElse(RemoteAddress.Unknown)
+
   // Parenthesis in the case matches below are required because otherwise scalafmt breaks.
   //noinspection ScalaUnnecessaryParentheses
   def all(implicit
       lc: LoggingContextOf[InstanceUUID],
       metrics: Metrics,
-  ): PartialFunction[HttpRequest, Http.IncomingConnection => Future[HttpResponse]] = {
+  ): PartialFunction[HttpRequest, Future[HttpResponse]] = {
     val apiMetrics = metrics.daml.HttpJsonApi
     type DispatchFun =
       PartialFunction[HttpRequest, LoggingContextOf[InstanceUUID with RequestID] => Future[
@@ -126,11 +138,11 @@ class Endpoints(
               )
         )
       // format: off
-        case req @ HttpRequest(GET,
-          Uri(_, _, Slash(Segment("v1", Slash(Segment("packages", Slash(Segment(packageId, Empty)))))), _, _),
-          _, _, _) =>
-            (implicit lc => Timed.future(apiMetrics.downloadPackageTimer, downloadPackage(req, packageId)))
-        // format: on
+      case req @ HttpRequest(GET,
+        Uri(_, _, Slash(Segment("v1", Slash(Segment("packages", Slash(Segment(packageId, Empty)))))), _, _),
+        _, _, _) =>
+          (implicit lc => Timed.future(apiMetrics.downloadPackageTimer, downloadPackage(req, packageId)))
+      // format: on
     }
     val liveOrHealthDispatch: DispatchFun = {
       case HttpRequest(GET, Uri.Path("/livez"), _, _, _) =>
@@ -147,17 +159,18 @@ class Endpoints(
       allocatePartyDispatch orElse
       packageManagementDispatch orElse
       liveOrHealthDispatch) &&& { case r => r }) andThen { case (lcFhr, req) =>
-      (connection: Http.IncomingConnection) =>
-        extendWithRequestIdLogCtx(implicit lc => {
-          val t0 = System.nanoTime
-          logger.info(s"Incoming request on ${req.uri} from ${connection.remoteAddress}")
-          metrics.daml.HttpJsonApi.httpRequestThroughput.mark()
-          for {
-            res <- lcFhr(lc)
-            _ = logger.trace(s"Processed request after ${System.nanoTime() - t0}ns")
-            _ = logger.info(s"Responding to client with HTTP ${res.status}")
-          } yield res
-        })
+      extendWithRequestIdLogCtx(implicit lc => {
+        val t0 = System.nanoTime
+        logger.info(s"Incoming request on ${req.uri} from ${requestSource(req)}")
+        metrics.daml.HttpJsonApi.httpRequestThroughput.mark()
+        for {
+          res <- lcFhr(lc)
+          _ = {
+            logger.trace(s"Processed request after ${System.nanoTime() - t0}ns")
+            logger.info(s"Responding to client with HTTP ${res.status}")
+          }
+        } yield res
+      })
     }
   }
 

diff --git a/ledger-service/http-json/src/main/scala/com/digitalasset/http/HttpService.scala b/ledger-service/http-json/src/main/scala/com/digitalasset/http/HttpService.scala
@@ -5,9 +5,7 @@ package com.daml.http
 
 import akka.actor.{ActorSystem, Cancellable}
 import akka.http.scaladsl.Http
-import akka.stream.scaladsl.Sink
 import akka.http.scaladsl.Http.ServerBinding
-import akka.http.scaladsl.model.HttpResponse
 import akka.http.scaladsl.settings.ServerSettings
 import akka.stream.Materializer
 import com.daml.auth.TokenHolder
@@ -195,16 +193,14 @@ object HttpService {
         websocketService,
       )
 
-      ignoreConParam = (res: Future[HttpResponse]) => (_: Http.IncomingConnection) => res
-
       defaultEndpoints =
         jsonEndpoints.all orElse
-          (websocketEndpoints.transactionWebSocket andThen ignoreConParam) orElse
-          (EndpointsCompanion.notFound andThen ignoreConParam)
+          websocketEndpoints.transactionWebSocket orElse
+          EndpointsCompanion.notFound
 
       allEndpoints = staticContentConfig.cata(
         c =>
-          (StaticContentEndpoints.all(c) andThen ignoreConParam) orElse
+          StaticContentEndpoints.all(c) orElse
             defaultEndpoints,
         defaultEndpoints,
       )
@@ -213,13 +209,7 @@ object HttpService {
         Http()
           .newServerAt(address, httpPort)
           .withSettings(settings)
-          .connectionSource()
-          .to {
-            Sink.foreach { connection =>
-              connection.handleWithAsyncHandler(allEndpoints(_)(connection))
-            }
-          }
-          .run()
+          .bind(allEndpoints)
       )
 
       _ <- either(portFile.cata(f => createPortFile(f, binding), \/-(()))): ET[Unit]