Add support for HA proxy protocol [MAB-25933]

README.md

@@ -75,3 +75,37 @@ for more details about how to contribute.
 Copyright 2010 Twitter, Inc.
 
 Licensed under the Apache License, Version 2.0: https://www.apache.org/licenses/LICENSE-2.0
+
+
+## Local build
+
+### Create working directory
+```
+mkdir -p <HOME>/runner/work/finagle/finagle
+cd <HOME>/runner/work/finagle/finagle
+```
+### Clone repository
+```
+git clone <this> .
+```
+### Build dependencies
+```
+mkdir -p <HOME>/runner/bin
+curl -sL -o <HOME>/runner/bin/dodo https://raw.githubusercontent.com/twitter/dodo/develop/bin/build
+chmod 755 <HOME>/runner/bin/dodo
+cd <HOME>/runner/work/finagle/finagle
+IVY_HOME=<HOME>/runner <HOME>/runner/bin/dodo --branch develop --no-test --publish-m2 --verbose finagle
+```
+### Build Finagle
+```
+cd <HOME>/runner/work/finagle/finagle
+./sbt update
+./sbt compile
+```
+
+## Local run tests
+
+Run the following to run tests for the most relevant Finagle HTTP packages.
+```
+./sbt "finagle-base-http/test" "finagle-http/test" "finagle-http2/test" "finagle-netty4/test" "finagle-netty4-http/test" "finagle-core/test"
+```

build.sbt

@@ -68,6 +68,7 @@ val netty4LibsTest = Seq(
 )
 val netty4Http = "io.netty" % "netty-codec-http" % netty4Version
 val netty4Http2 = "io.netty" % "netty-codec-http2" % netty4Version
+val netty4HAProxy = "io.netty" % "netty-codec-haproxy" % netty4Version
 val opencensusVersion = "0.24.0"
 val jacksonVersion = "2.13.4"
 val jacksonLibs = Seq(
@@ -430,7 +431,8 @@ lazy val finagleNetty4 = Project(
       util("core"),
       util("codec"),
       util("lint"),
-      util("stats")
+      util("stats"),
+      netty4HAProxy
     ) ++ netty4Libs
   ).dependsOn(
     finagleCore % "compile->compile;test->test",

finagle-base-http/src/main/scala/com/twitter/finagle/http/Request.scala

@@ -216,6 +216,26 @@ abstract class Request private extends Message {
   def remotePort: Int =
     remoteSocketAddress.getPort
 
+  /**
+   * The client source address of the request.
+   *
+   * Typically HTTP server is placed behind load balancer and therefore
+   * [[remoteAddress]] is pointing to it. In such cases client source address
+   * is lost.
+
+   * @note Value will be fulfilled only if:
+   *       - HA proxy protocol is enabled on finagle HTTP server
+   *       - Load balancer pre-append request with proxy protocol message v1|v2
+   */
+  def clientSourceAddress: Option[InetAddress]
+
+  /**
+   * The client destination port of the request.
+   *
+   * @see [[clientSourceAddress]] for additional information.
+   */
+  def clientDestinationPort: Option[Int]
+
   // The get*Param methods below are for Java compatibility.  Note Scala default
   // arguments aren't compatible with Java, so we need two versions of each.
 
@@ -468,6 +488,8 @@ object Request {
 
     def ctx: Schema.Record = request.ctx
     def remoteSocketAddress: InetSocketAddress = request.remoteSocketAddress
+    def clientSourceAddress: Option[InetAddress] = None
+    def clientDestinationPort: Option[Int] = None
     def chunkReader: Reader[Chunk] = request.chunkReader
     def chunkWriter: Writer[Chunk] = request.chunkWriter
     override lazy val cookies: CookieMap = request.cookies
@@ -495,8 +517,10 @@ object Request {
   private[finagle] final class Inbound(
     val chunkReader: Reader[Chunk],
     val remoteSocketAddress: InetSocketAddress,
-    val trailers: HeaderMap)
-      extends Request {
+    val trailers: HeaderMap,
+    val clientSourceAddress: Option[InetAddress] = None,
+    val clientDestinationPort: Option[Int] = None
+  ) extends Request {
 
     def chunkWriter: Writer[Chunk] = FailingWriter
 
@@ -531,8 +555,7 @@ object Request {
   private[finagle] final class Impl(
     val chunkReader: Reader[Chunk],
     val chunkWriter: Writer[Chunk],
-    val remoteSocketAddress: InetSocketAddress)
-      extends Request {
+    val remoteSocketAddress: InetSocketAddress) extends Request {
 
     def this(chunkReader: Reader[Chunk], remoteSocketAddress: InetSocketAddress) =
       this(chunkReader, FailingWriter, remoteSocketAddress)
@@ -562,5 +585,8 @@ object Request {
     def uri_=(uri: String): Unit = {
       _uri = uri
     }
+
+    def clientSourceAddress: Option[InetAddress] = None
+    def clientDestinationPort: Option[Int] = None
   }
 }

finagle-base-http/src/main/scala/com/twitter/finagle/http/param/params.scala

@@ -89,6 +89,30 @@ object Streaming {
     Enabled(fixedLengthStreamedAfter)
 }
 
+sealed abstract class HAProxyProtocol private {
+  def enabled: Boolean
+  final def disabled: Boolean = !enabled
+}
+
+object HAProxyProtocol {
+
+  implicit val haProxyProtocolParam: Stack.Param[HAProxyProtocol] =
+    Stack.Param(Disabled)
+
+  private[finagle] case object Enabled extends HAProxyProtocol {
+    def enabled: Boolean = true
+  }
+
+  private[finagle] case object Disabled extends HAProxyProtocol {
+    def enabled: Boolean = false
+  }
+
+  def apply(enabled: Boolean): HAProxyProtocol = {
+    if (enabled) Enabled
+    else Disabled
+  }
+}
+
 case class FixedLengthStreamedAfter(size: StorageUnit)
 object FixedLengthStreamedAfter {
   implicit val fixedLengthStreamedAfter: Stack.Param[FixedLengthStreamedAfter] =

finagle-core/src/main/scala/com/twitter/finagle/transport/TransportContext.scala

@@ -2,7 +2,7 @@ package com.twitter.finagle.transport
 
 import com.twitter.finagle.ssl.session.{NullSslSessionInfo, SslSessionInfo}
 import com.twitter.util.Updatable
-import java.net.SocketAddress
+import java.net.{InetAddress, SocketAddress}
 
 /**
  * Exposes a way to control the transport, and read off properties from the
@@ -26,6 +26,37 @@ abstract class TransportContext {
    * @note If SSL/TLS is not being used a `NullSslSessionInfo` will be returned instead.
    */
   def sslSessionInfo: SslSessionInfo
+
+  /**
+   * Initial, typically client, source address of this transport.
+   *
+   * Typically HTTP server is placed behind load balancer and initial transport protocol
+   * information is therefore lost.
+   *
+   * @note Property is fulfilled only if HAProxyProtocol message is received through channel.
+   */
+  def clientSourceAddress: Option[InetAddress] = None
+
+  /**
+   * Initial, typically client, source port of this transport.
+   *
+   * @see For more information check [[clientSourceAddress]]
+   */
+  def clientSourcePort: Option[Int] = None
+
+  /**
+   * Initial, typically client, destination address of this transport.
+   *
+   * @see For more information check [[clientSourceAddress]]
+   */
+  def clientDestinationAddress: Option[InetAddress] = None
+
+  /**
+   * Initial, typically client, destination port of this transport.
+   *
+   * @see For more information check [[clientSourceAddress]]
+   */
+  def clientDestinationPort: Option[Int] = None
 }
 
 private[finagle] class SimpleTransportContext(
@@ -46,4 +77,8 @@ private[finagle] class UpdatableContext(first: TransportContext)
   def localAddress: SocketAddress = underlying.localAddress
   def remoteAddress: SocketAddress = underlying.remoteAddress
   def sslSessionInfo: SslSessionInfo = underlying.sslSessionInfo
+  override def clientSourceAddress: Option[InetAddress] = underlying.clientSourceAddress
+  override def clientSourcePort: Option[Int] = underlying.clientSourcePort
+  override def clientDestinationAddress: Option[InetAddress] = underlying.clientDestinationAddress
+  override def clientDestinationPort: Option[Int] = underlying.clientDestinationPort
 }

finagle-http/src/main/scala/com/twitter/finagle/Http.scala

@@ -7,8 +7,7 @@ import com.twitter.finagle.http._
 import com.twitter.finagle.http.codec.HttpServerDispatcher
 import com.twitter.finagle.http.StreamTransport
 import com.twitter.finagle.http.filter._
-import com.twitter.finagle.http.param.ClientKerberosConfiguration
-import com.twitter.finagle.http.param.ServerKerberosConfiguration
+import com.twitter.finagle.http.param.{ClientKerberosConfiguration, HAProxyProtocol, ServerKerberosConfiguration}
 import com.twitter.finagle.http.service.HttpResponseClassifier
 import com.twitter.finagle.http2.Http2Listener
 import com.twitter.finagle.netty4.http.Netty4HttpListener
@@ -68,21 +67,24 @@ object Http extends Client[Request, Response] with HttpRichClient with Server[Re
   private[finagle] final class HttpImpl private (
     private[finagle] val clientEndpointer: Stackable[ServiceFactory[Request, Response]],
     private[finagle] val serverTransport: Transport[Any, Any] => StreamTransport[Response, Request],
-    private[finagle] val listener: Stack.Params => Listener[Any, Any, TransportContext])
+    private[finagle] val listener: Stack.Params => Listener[Any, Any, TransportContext],
+    private[finagle] val protocol: Protocol)
 
   private[finagle] object HttpImpl {
     implicit val httpImplParam: Stack.Param[HttpImpl] = Stack.Param(Http11Impl)
 
     val Http11Impl: Http.HttpImpl = new Http.HttpImpl(
       ClientEndpointer.HttpEndpointer,
       new Netty4ServerStreamTransport(_),
-      Netty4HttpListener
+      Netty4HttpListener,
+      Protocol.HTTP_1_1
     )
 
     val Http2Impl: Http.HttpImpl = new Http.HttpImpl(
       ClientEndpointer.Http2Endpointer,
       new Netty4ServerStreamTransport(_),
-      Http2Listener.apply _
+      Http2Listener.apply _,
+      Protocol.HTTP_2
     )
   }
 
@@ -497,6 +499,35 @@ object Http extends Client[Request, Response] with HttpRichClient with Server[Re
     def withStreaming(fixedLengthStreamedAfter: StorageUnit): Server =
       configured(http.param.Streaming(fixedLengthStreamedAfter))
 
+    /**
+     * Enable/disable HA proxy protocol v1|v2
+     *
+     * Typically HTTP server is placed behind load balancer and therefore
+     * remote address of the request is pointing to it. In such case original
+     * client source address and destination port of the request is lost.
+     *
+     * Enable HA proxy protocol only if load balancer supports and pre-append requests
+     * with proxy protocol message v1|v2.
+     *
+     * @note Handling HA proxy protocol is currently NOT supported for:
+     *       - HTTP protocol version 2
+     *       - HTTP1.1 streaming
+     *
+     * @note Enabling HA Proxy protocol will disable HTTP2 and disable streaming for HTTP1.1
+     *
+     * @see [[https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt]]
+     */
+    def withHAProxyProtocol(enabled: Boolean): Server = {
+      if (enabled) {
+        this
+          .withNoHttp2
+          .withStreaming(false)
+          .configured(http.param.HAProxyProtocol(enabled))
+      } else {
+        configured(http.param.HAProxyProtocol(enabled))
+      }
+    }
+
     /**
      * Enables decompression of http content bodies.
      */
@@ -633,8 +664,24 @@ object Http extends Client[Request, Response] with HttpRichClient with Server[Re
             case Protocol.HTTP_1_1 => withNoHttp2
           }
 
+      validate()
+
       server.superServe(addr, factory)
     }
+
+    /**
+     * HTTP server configuration validation
+     *
+     * @see [[com.twitter.finagle.Http.Server.withHAProxyProtocol]]
+     */
+    private def validate(): Unit = {
+      if (params.contains[HAProxyProtocol] && params[HAProxyProtocol].enabled) {
+        if (
+          (params.contains[HttpImpl] && params[HttpImpl].protocol == Protocol.HTTP_2) ||
+          (params.contains[http.param.Streaming] && params[http.param.Streaming].enabled)
+        ) throw new Exception("Validation of HTTP server with HA proxy protocol enabled failed.")
+      }
+    }
   }
 
   def server: Http.Server = Server()