Implement reverseProxy directive (#240)

CONTRIBUTING.md

@@ -199,7 +199,7 @@ All documentation must abide by the following maxims:
 - When renaming Markdown files, add a rewrite rule to the `.htaccess` file to not break external links.
 
 All documentation is preferred to be in Lightbend's standard documentation format [Paradox](https://github.com/lightbend/paradox).
-The language used by Paradox is a super-set or Markdown which supports most Github Flavored Markdown extensions as well as additional directives to facilitate writing documentation for software projects.
+The language used by Paradox is a super-set of Markdown which supports most Github Flavored Markdown extensions as well as additional directives to facilitate writing documentation for software projects.
 Refer to its documentation to learn about the more advanced features it provides (including code etc).
 
 To generate documentation you can:

akka-http-core/src/main/scala/akka/http/scaladsl/model/headers/ViaIntermediary.scala

@@ -0,0 +1,15 @@
+/*
+ * Copyright (C) 2019 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.http.scaladsl.model.headers
+
+import akka.http.impl.util.{ Rendering, ValueRenderable }
+
+final case class ViaIntermediary(protocolName: Option[String], protocolVersion: String, receivedBy: String) extends ValueRenderable {
+  def render[R <: Rendering](r: R): r.type = {
+    protocolName.foreach(name => r ~~ s"$name/")
+    r ~~ protocolVersion ~~ ' ' ~~ receivedBy
+    r
+  }
+}

akka-http-core/src/main/scala/akka/http/scaladsl/model/headers/headers.scala

@@ -1050,6 +1050,20 @@ final case class `User-Agent`(products: immutable.Seq[ProductVersion]) extends j
   def getProducts: Iterable[jm.headers.ProductVersion] = products.asJava
 }
 
+// https://tools.ietf.org/html/rfc7230#section-5.7.1
+object Via extends ModeledCompanion[Via]
+final case class Via(intermediaries: immutable.Seq[ViaIntermediary]) extends RequestResponseHeader {
+  def renderValue[R <: Rendering](r: R): r.type = intermediaries match {
+    case Seq() => r
+    case i +: is =>
+      r ~~ i
+      is.foreach(r ~~ ", " ~~ _)
+      r
+  }
+
+  protected def companion = Via
+}
+
 // http://tools.ietf.org/html/rfc7235#section-4.1
 object `WWW-Authenticate` extends ModeledCompanion[`WWW-Authenticate`] {
   def apply(first: HttpChallenge, more: HttpChallenge*): `WWW-Authenticate` = apply(immutable.Seq(first +: more: _*))

akka-http-tests/src/test/scala/akka/http/scaladsl/server/directives/ReverseProxyDirectivesSpec.scala

@@ -0,0 +1,82 @@
+/*
+ * Copyright (C) 2019 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.http.scaladsl.server.directives
+
+import java.net.InetAddress
+
+import scala.concurrent.Future
+import scala.concurrent.duration.Duration
+
+import akka.http.scaladsl.model._
+import akka.http.scaladsl.model.headers._
+import akka.http.scaladsl.server.RoutingSpec
+import akka.http.scaladsl.server.directives.ReverseProxyDirectives._
+
+class ReverseProxyDirectivesSpec extends RoutingSpec {
+  "The reverseProxy directive" should {
+    "forward requests to the configured target" in new ReverseProxyDirectiveSpecHelper {
+
+      val targetConfig = ReverseProxyTargetConfig("https://target.domain:1234", false)
+
+      Get("http://notthetarget.domain/the/path?foo=bar&bar=baz") ~> reverseProxy(magnet) ~> check {
+        status shouldBe StatusCodes.OK
+        receivedRequest should not be null
+        receivedRequest.effectiveUri(false) shouldBe Uri("https://target.domain:1234/the/path?foo=bar&bar=baz")
+      }
+    }
+
+    "use unmatched the unmatched path as configured" in new ReverseProxyDirectiveSpecHelper {
+
+      val targetConfig = ReverseProxyTargetConfig("https://target.domain:1234", useUnmatchedPath = true)
+
+      Get("http://notthetarget.domain/the/path?foo=bar&bar=baz") ~> pathPrefix("the")(reverseProxy(magnet)) ~> check {
+        status shouldBe StatusCodes.OK
+        receivedRequest should not be null
+        receivedRequest.effectiveUri(false) shouldBe Uri("https://target.domain:1234/path?foo=bar&bar=baz")
+      }
+    }
+
+    "strip hop-by-hop headers" in new ReverseProxyDirectiveSpecHelper {
+
+      val targetConfig = ReverseProxyTargetConfig("https://target.domain:1234", false)
+
+      val request = Get("http://notthetarget.domain/the/path?foo=bar&bar=baz")
+        .withHeaders(
+          `X-Real-Ip`(RemoteAddress(InetAddress.getLocalHost)),
+          Connection("close"),
+          `Proxy-Authenticate`(HttpChallenge("scheme", "realm")),
+          `Proxy-Authorization`(BasicHttpCredentials("username", "password")),
+          `Transfer-Encoding`(TransferEncodings.chunked),
+          Upgrade(List(UpgradeProtocol("protocol")))
+        )
+
+      request ~> withRequestTimeout(Duration.Inf)(reverseProxy(magnet)) ~> check {
+        status shouldBe StatusCodes.OK
+        receivedRequest should not be null
+        receivedRequest.header[`Timeout-Access`] should not be defined
+        receivedRequest.header[Connection] should not be defined
+        receivedRequest.header[`Proxy-Authenticate`] should not be defined
+        receivedRequest.header[`Proxy-Authorization`] should not be defined
+        receivedRequest.header[`Transfer-Encoding`] should not be defined
+        receivedRequest.header[Upgrade] should not be defined
+      }
+    }
+  }
+}
+
+trait ReverseProxyDirectiveSpecHelper {
+  var receivedRequest: HttpRequest = null
+
+  def targetConfig: ReverseProxyTargetConfig
+
+  val magnet = new ReverseProxyTargetMagnet {
+    def config = targetConfig
+
+    val httpClient: HttpRequest => Future[HttpResponse] = request => {
+      receivedRequest = request
+      Future.successful(HttpResponse())
+    }
+  }
+}

akka-http/src/main/scala/akka/http/javadsl/server/Directives.scala

@@ -7,11 +7,11 @@ package akka.http.javadsl.server
 import java.util.function.{ BiFunction, Function, Supplier }
 
 import akka.annotation.ApiMayChange
-import akka.http.javadsl.server.directives.FramedEntityStreamingDirectives
+import akka.http.javadsl.server.directives.ReverseProxyDirectives
 
 import scala.annotation.varargs
 
-abstract class AllDirectives extends FramedEntityStreamingDirectives
+abstract class AllDirectives extends ReverseProxyDirectives
 
 /**
  * Collects all default directives into one class for simple importing of static functions.

akka-http/src/main/scala/akka/http/javadsl/server/directives/ReverseProxyDirectives.scala

@@ -0,0 +1,42 @@
+/*
+ * Copyright (C) 2019 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.http.javadsl.server.directives
+
+import java.util.concurrent.CompletionStage
+import java.util.function.{ Function => JFunction }
+
+import akka.annotation.ApiMayChange
+import akka.http.javadsl.model.{ Uri, HttpRequest => JHttpRequest, HttpResponse => JHttpResponse }
+import akka.http.javadsl.server.Route
+import akka.http.scaladsl.model
+import akka.http.scaladsl.model.HttpResponse
+import akka.http.scaladsl.server.directives.ReverseProxyDirectives.{ ReverseProxyTargetConfig, ReverseProxyTargetMagnet }
+import akka.http.scaladsl.server.{ Directives => D }
+
+import scala.compat.java8.FunctionConverters._
+import scala.compat.java8.FutureConverters._
+import scala.concurrent.Future
+
+abstract class ReverseProxyDirectives extends FramedEntityStreamingDirectives {
+  import akka.http.impl.util.JavaMapping.Implicits._
+
+  private val javaToScalaResponse = ((res: JHttpResponse) => res.asScala).asJava
+
+  @ApiMayChange
+  def reverseProxy(
+    targetUri:        Uri,
+    useUnmatchedPath: Boolean,
+    httpClient:       JFunction[JHttpRequest, CompletionStage[JHttpResponse]]
+  ): Route = {
+    def _client = httpClient
+    RouteAdapter(
+      D.reverseProxy(new ReverseProxyTargetMagnet {
+        val config = ReverseProxyTargetConfig(targetUri.asScala, useUnmatchedPath)
+        val httpClient: model.HttpRequest => Future[model.HttpResponse] =
+          req => _client(req).thenApply[HttpResponse](javaToScalaResponse).toScala
+      })
+    )
+  }
+}

akka-http/src/main/scala/akka/http/scaladsl/server/Directives.scala

@@ -37,6 +37,7 @@ trait Directives extends RouteConcatenation
   with SecurityDirectives
   with WebSocketDirectives
   with FramedEntityStreamingDirectives
+  with ReverseProxyDirectives
 
 /**
  * Collects all default directives into one object for simple importing.

akka-http/src/main/scala/akka/http/scaladsl/server/directives/ReverseProxyDirectives.scala

@@ -0,0 +1,151 @@
+/*
+ * Copyright (C) 2019 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.http.scaladsl.server.directives
+
+import scala.concurrent.Future
+import scala.util.Success
+import akka.actor.ActorSystem
+import akka.annotation.ApiMayChange
+import akka.http.scaladsl.Http
+import akka.http.scaladsl.model.Uri.Authority
+import akka.http.scaladsl.model._
+import akka.http.scaladsl.model.headers._
+import akka.http.scaladsl.server.Directives._
+import akka.http.scaladsl.server.Route
+import akka.http.scaladsl.server.directives.ReverseProxyDirectives.{ ReverseProxyTargetConfig, ReverseProxyTargetMagnet }
+import akka.util.ByteString
+
+trait ReverseProxyDirectives {
+
+  @ApiMayChange
+  def reverseProxy(target: ReverseProxyTargetMagnet): Route =
+    extractExecutionContext { implicit ec =>
+      extractMaterializer { implicit mat =>
+        // todo customize response
+        // drain response entity if incoming request timeout occurs
+        def processTimeout(proxyResult: Future[HttpResponse]): HttpRequest => HttpResponse = _ => {
+          proxyResult.andThen {
+            case Success(proxyResponse) => proxyResponse.discardEntityBytes()
+          }
+
+          HttpResponse(StatusCodes.ServiceUnavailable, entity =
+            HttpEntity.Strict(
+              ContentTypes.`application/json`,
+              ByteString("""{"error":"The server was not able to produce a timely response to your request."}""")
+            )
+          )
+        }
+
+        extractRequestContext { ctx =>
+          // we don't need to use request.effectiveUri here since we're going to overwrite the scheme and authority
+          val incomingUri = ctx.request.uri
+          val outgoingUri =
+            (if (target.config.useUnmatchedPath) incomingUri.withPath(ctx.unmatchedPath) else incomingUri)
+              .withAuthority(target.config.targetAuthority)
+              .withScheme(target.config.targetScheme)
+
+          val outgoingRequest = mapProxyRequest(ctx.request, target.config).withUri(outgoingUri)
+          val eventualResponse = target.httpClient(outgoingRequest)
+
+          withRequestTimeoutResponse(processTimeout(eventualResponse)) {
+            complete(eventualResponse)
+          }
+        }
+      }
+    }
+
+  // remove any headers that shouldn't be passed through a proxy
+  private def mapProxyRequest(request: HttpRequest, config: ReverseProxyTargetConfig): HttpRequest = {
+    val incomingHeaders = request.headers
+
+    val remoteAddressOption = request.header[`Remote-Address`].map(_.address)
+    val xForwardedForAddressesOption = request.header[`X-Forwarded-For`].map(_.addresses)
+
+    val updatedXRealIpHeaderOption = request.header[`X-Real-Ip`]
+      .map(_.address)
+      .orElse(xForwardedForAddressesOption.flatMap(_.headOption))
+      .orElse(remoteAddressOption)
+      .map(`X-Real-Ip`(_))
+
+    val updatedXForwardedForHeaderOption = remoteAddressOption
+      .map(_ +: xForwardedForAddressesOption.getOrElse(Nil))
+      .map(`X-Forwarded-For`(_))
+
+    val intermediary = {
+      val (protocol, version) = {
+        val parts = request.protocol.value.split('/')
+        if (parts(0) == "HTTP" && parts.length > 1) None -> parts(1)
+        else if (parts.length > 1) Some(parts(0)) -> parts(1)
+        else None -> parts(0)
+      }
+
+      config.viaId.map(ViaIntermediary(protocol, version, _))
+    }
+
+    val maybeVia = request.header[Via]
+      .map(via => via.copy(intermediaries = via.intermediaries ++ intermediary))
+      .orElse(intermediary.map(i => Via(List(i))))
+
+    val outgoingHeaders = incomingHeaders.flatMap {
+      // region hop-by-hop headers https://tools.ietf.org/html/rfc2616#section-13.5.1
+      case _: Connection            => Nil
+      // keep alive header is not included in modeled headers
+      case _: `Proxy-Authenticate`  => Nil
+      case _: `Proxy-Authorization` => Nil
+      case h if h.is("te")          => Nil
+      case h if h.is("trailers")    => Nil
+      case _: `Transfer-Encoding`   => Nil
+      case _: Upgrade               => Nil
+      // endregion
+      case _: `X-Real-Ip`           => updatedXRealIpHeaderOption
+      case _: `X-Forwarded-For`     => updatedXForwardedForHeaderOption
+      case _: `Timeout-Access`      => Nil
+      case _: Via                   => Nil // added back at the end
+      case h                        => Some(h)
+    } ++ maybeVia
+
+    request.withHeaders(outgoingHeaders)
+  }
+}
+
+object ReverseProxyDirectives extends ReverseProxyDirectives {
+
+  trait ReverseProxyTargetConfig {
+    def targetScheme: String
+    def targetAuthority: Authority
+    def useUnmatchedPath: Boolean
+    def viaId: Option[String]
+  }
+
+  object ReverseProxyTargetConfig {
+    def apply(baseUri: Uri, useUnmatchedPath: Boolean): ReverseProxyTargetConfig = ReverseProxyTargetConfigImpl(baseUri, useUnmatchedPath)
+  }
+
+  private case class ReverseProxyTargetConfigImpl(
+    baseUri:          Uri,
+    useUnmatchedPath: Boolean,
+    viaId:            Option[String] = None
+  ) extends ReverseProxyTargetConfig {
+    val targetScheme = baseUri.scheme
+    val targetAuthority = baseUri.authority
+  }
+
+  trait ReverseProxyTargetMagnet {
+    def config: ReverseProxyTargetConfig
+    def httpClient: HttpRequest => Future[HttpResponse]
+  }
+
+  object ReverseProxyTargetMagnet {
+    import scala.language.implicitConversions
+
+    implicit def fromConfig(targetConfig: ReverseProxyTargetConfig)(implicit system: ActorSystem) =
+      new ReverseProxyTargetMagnet {
+        val config = targetConfig
+        val httpClient: HttpRequest => Future[HttpResponse] = Http().singleRequest(_)
+      }
+
+    implicit def fromUri(uri: Uri)(implicit system: ActorSystem) = fromConfig(ReverseProxyTargetConfigImpl(uri, false))
+  }
+}