[KYUUBI #3052][FOLLOWUP] Do not use the ip in proxy http header for authentication to prevent CVE

### _Why are the changes needed?_

Because the client can specify any ip in http header, to prevent CVE issue, we do not use it for authentication.

### _How was this patch tested?_
- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [x] [Run test](https://kyuubi.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request

Closes #3078 from turboFei/http_real_ip_cve.

Closes #3052

e7c41ea [Fei Wang] prevent cve

Authored-by: Fei Wang <fwang12@ebay.com>
Signed-off-by: Fei Wang <fwang12@ebay.com>

Author: turboFei

docs/deployment/settings.md

@@ -273,7 +273,7 @@ kyuubi.frontend.mysql.min.worker.threads|9|Minimum number of threads in the comm
 kyuubi.frontend.mysql.netty.worker.threads|<undefined>|Number of thread in the netty worker event loop of MySQL frontend service. Use min(cpu_cores, 8) in default.|int|1.4.0
 kyuubi.frontend.mysql.worker.keepalive.time|PT1M|Time(ms) that an idle async thread of the command execution thread pool will wait for a new task to arrive before terminating in MySQL frontend service|duration|1.4.0
 kyuubi.frontend.protocols|THRIFT_BINARY|A comma separated list for all frontend protocols <ul> <li>THRIFT_BINARY - HiveServer2 compatible thrift binary protocol.</li> <li>THRIFT_HTTP - HiveServer2 compatible thrift http protocol.</li> <li>REST - Kyuubi defined REST API(experimental).</li>  <li>MYSQL - MySQL compatible text protocol(experimental).</li> </ul>|seq|1.4.0
-kyuubi.frontend.proxy.http.client.ip.header|X-Real-IP|The http header to record the real client ip address. If your server is behind a load balancer or other proxy, the server will see this load balancer or proxy IP address as the client IP address, to get around this common issue, most load balancers or proxies offer the ability to record the real remote IP address in an HTTP herader that will be added to the request for other devices to use.|string|1.6.0
+kyuubi.frontend.proxy.http.client.ip.header|X-Real-IP|The http header to record the real client ip address. If your server is behind a load balancer or other proxy, the server will see this load balancer or proxy IP address as the client IP address, to get around this common issue, most load balancers or proxies offer the ability to record the real remote IP address in an HTTP header that will be added to the request for other devices to use. Note that, because the header value can be specified to any ip address, so it will not be used for authentication.|string|1.6.0
 kyuubi.frontend.rest.bind.host|&lt;undefined&gt;|Hostname or IP of the machine on which to run the REST frontend service.|string|1.4.0
 kyuubi.frontend.rest.bind.port|10099|Port of the machine on which to run the REST frontend service.|int|1.4.0
 kyuubi.frontend.thrift.backoff.slot.length|PT0.1S|Time to back off during login to the thrift frontend service.|duration|1.4.0

kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala

@@ -585,8 +585,9 @@ object KyuubiConf {
       .doc("The http header to record the real client ip address. If your server is behind a load" +
         " balancer or other proxy, the server will see this load balancer or proxy IP address as" +
         " the client IP address, to get around this common issue, most load balancers or proxies" +
-        " offer the ability to record the real remote IP address in an HTTP herader that will be" +
-        " added to the request for other devices to use.")
+        " offer the ability to record the real remote IP address in an HTTP header that will be" +
+        " added to the request for other devices to use. Note that, because the header value can" +
+        " be specified to any ip address, so it will not be used for authentication.")
       .version("1.6.0")
       .stringConf
       .createWithDefault("X-Real-IP")

kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiRestFrontendService.scala

@@ -173,9 +173,16 @@ class KyuubiRestFrontendService(override val serverable: Serverable)
   }
 
   def getUserName(sessionConf: Map[String, String]): String = {
+    // using the remote ip address instead of that in proxy http header for authentication
+    val ipAddress = AuthenticationFilter.getUserIpAddress
     val realUser: String = ServiceUtils.getShortName(
       Option(AuthenticationFilter.getUserName).filter(_.nonEmpty).getOrElse("anonymous"))
-    getProxyUser(sessionConf, Option(AuthenticationFilter.getUserIpAddress).orNull, realUser)
+    getProxyUser(sessionConf, ipAddress, realUser)
+  }
+
+  def getIpAddress: String = {
+    Option(AuthenticationFilter.getUserProxyHeaderIpAddress).getOrElse(
+      AuthenticationFilter.getUserIpAddress)
   }
 
   private def getProxyUser(

kyuubi-server/src/main/scala/org/apache/kyuubi/server/KyuubiTHttpFrontendService.scala

@@ -261,14 +261,16 @@ final class KyuubiTHttpFrontendService(
   }
 
   override protected def getIpAddress: String = {
-    SessionManager.getIpAddress
+    Option(SessionManager.getProxyHttpHeaderIpAddress).getOrElse(SessionManager.getIpAddress)
   }
 
   override protected def getUserName(req: TOpenSessionReq): String = {
     var userName: String = SessionManager.getUserName
+    // using the remote ip address instead of that in proxy http header for authentication
+    val ipAddress: String = SessionManager.getIpAddress
     if (userName == null) userName = req.getUsername
     userName = getShortName(userName)
-    val effectiveClientUser: String = getProxyUser(req.getConfiguration, getIpAddress, userName)
+    val effectiveClientUser: String = getProxyUser(req.getConfiguration, ipAddress, userName)
     debug("Client's username: " + effectiveClientUser)
     effectiveClientUser
   }

kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/AdminResource.scala

@@ -27,7 +27,6 @@ import io.swagger.v3.oas.annotations.tags.Tag
 import org.apache.kyuubi.{Logging, Utils}
 import org.apache.kyuubi.server.KyuubiServer
 import org.apache.kyuubi.server.api.ApiRequestContext
-import org.apache.kyuubi.server.http.authentication.AuthenticationFilter
 
 @Tag(name = "Admin")
 @Produces(Array(MediaType.APPLICATION_JSON))
@@ -44,7 +43,7 @@ private[v1] class AdminResource extends ApiRequestContext with Logging {
   @Path("refresh/hadoop_conf")
   def refreshFrontendHadoopConf(): Response = {
     val userName = fe.getUserName(Map.empty)
-    val ipAddress = AuthenticationFilter.getUserIpAddress
+    val ipAddress = fe.getIpAddress
     info(s"Receive refresh Kyuubi server hadoop conf request from $userName/$ipAddress")
     if (!userName.equals(administrator)) {
       throw new NotAllowedException(

kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/BatchesResource.scala

@@ -38,7 +38,6 @@ import org.apache.kyuubi.engine.ApplicationOperation._
 import org.apache.kyuubi.operation.{BatchJobSubmission, FetchOrientation, OperationState}
 import org.apache.kyuubi.server.api.ApiRequestContext
 import org.apache.kyuubi.server.api.v1.BatchesResource._
-import org.apache.kyuubi.server.http.authentication.AuthenticationFilter
 import org.apache.kyuubi.server.metadata.MetadataManager
 import org.apache.kyuubi.server.metadata.api.Metadata
 import org.apache.kyuubi.service.authentication.KyuubiAuthenticationFactory
@@ -168,7 +167,7 @@ private[v1] class BatchesResource extends ApiRequestContext with Logging {
     request.setBatchType(request.getBatchType.toUpperCase(Locale.ROOT))
 
     val userName = fe.getUserName(request.getConf.asScala.toMap)
-    val ipAddress = AuthenticationFilter.getUserIpAddress
+    val ipAddress = fe.getIpAddress
     request.setConf(
       (request.getConf.asScala ++ Map(KYUUBI_CLIENT_IP_KEY -> ipAddress)).asJava)
     val sessionHandle = sessionManager.openBatchSession(

kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/SessionsResource.scala

@@ -35,7 +35,6 @@ import org.apache.kyuubi.client.api.v1.dto._
 import org.apache.kyuubi.events.KyuubiEvent
 import org.apache.kyuubi.operation.OperationHandle
 import org.apache.kyuubi.server.api.ApiRequestContext
-import org.apache.kyuubi.server.http.authentication.AuthenticationFilter
 import org.apache.kyuubi.session.KyuubiSession
 import org.apache.kyuubi.session.SessionHandle
 
@@ -142,7 +141,7 @@ private[v1] class SessionsResource extends ApiRequestContext with Logging {
   @Consumes(Array(MediaType.APPLICATION_JSON))
   def openSession(request: SessionOpenRequest): dto.SessionHandle = {
     val userName = fe.getUserName(request.getConfigs.asScala.toMap)
-    val ipAddress = AuthenticationFilter.getUserIpAddress
+    val ipAddress = fe.getIpAddress
     val handle = fe.be.openSession(
       TProtocolVersion.findByValue(request.getProtocolVersion),
       userName,

kyuubi-server/src/main/scala/org/apache/kyuubi/server/http/ThriftHttpServlet.scala

@@ -79,7 +79,6 @@ class ThriftHttpServlet(
   @throws[IOException]
   override def doPost(request: HttpServletRequest, response: HttpServletResponse): Unit = {
     var clientUserName: String = null
-    var clientIpAddress: String = null
     var requireNewCookie: Boolean = false
     try {
       if (conf.get(KyuubiConf.FRONTEND_THRIFT_HTTP_XSRF_FILTER_ENABLED)) {
@@ -119,12 +118,16 @@ class ThriftHttpServlet(
       val doAsQueryParam = getDoAsQueryParam(request.getQueryString)
       if (doAsQueryParam != null) SessionManager.setProxyUserName(doAsQueryParam)
 
-      clientIpAddress = Option(request.getHeader(conf.get(FRONTEND_PROXY_HTTP_CLIENT_IP_HEADER)))
-        .getOrElse(request.getRemoteAddr)
+      val clientIpAddress = request.getRemoteAddr
       debug("Client IP Address: " + clientIpAddress)
-      // Set the thread local ip address
       SessionManager.setIpAddress(clientIpAddress)
 
+      Option(request.getHeader(conf.get(FRONTEND_PROXY_HTTP_CLIENT_IP_HEADER))).foreach {
+        ipAddress =>
+          debug("Proxy Http Header Client IP Address: " + ipAddress)
+          SessionManager.setProxyHttpHeaderIpAddress(ipAddress)
+      }
+
       val forwarded_for = request.getHeader(X_FORWARDED_FOR_HEADER)
       if (forwarded_for != null) {
         debug(X_FORWARDED_FOR_HEADER + ":" + forwarded_for)
@@ -157,6 +160,7 @@ class ThriftHttpServlet(
       // Clear the thread locals
       SessionManager.clearUserName()
       SessionManager.clearIpAddress()
+      SessionManager.clearProxyHttpHeaderIpAddress()
       SessionManager.clearProxyUserName()
       SessionManager.clearForwardedAddresses()
     }

kyuubi-server/src/main/scala/org/apache/kyuubi/server/http/authentication/AuthenticationFilter.scala

@@ -114,8 +114,9 @@ class AuthenticationFilter(conf: KyuubiConf) extends Filter with Logging {
         HttpServletResponse.SC_UNAUTHORIZED,
         s"No auth scheme matched for $authorization")
     } else {
-      HTTP_CLIENT_IP_ADDRESS.set(Option(httpRequest.getHeader(
-        conf.get(FRONTEND_PROXY_HTTP_CLIENT_IP_HEADER))).getOrElse(httpRequest.getRemoteAddr))
+      HTTP_CLIENT_IP_ADDRESS.set(httpRequest.getRemoteAddr)
+      HTTP_PROXY_HEADER_CLIENT_IP_ADDRESS.set(
+        httpRequest.getHeader(conf.get(FRONTEND_PROXY_HTTP_CLIENT_IP_HEADER)))
       try {
         val authUser = matchedHandler.authenticate(httpRequest, httpResponse)
         if (authUser != null) {
@@ -126,6 +127,7 @@ class AuthenticationFilter(conf: KyuubiConf) extends Filter with Logging {
         case e: AuthenticationException =>
           HTTP_CLIENT_USER_NAME.remove()
           HTTP_CLIENT_IP_ADDRESS.remove()
+          HTTP_PROXY_HEADER_CLIENT_IP_ADDRESS.remove()
           httpResponse.setStatus(HttpServletResponse.SC_FORBIDDEN)
           httpResponse.sendError(HttpServletResponse.SC_FORBIDDEN, e.getMessage)
       }
@@ -163,11 +165,16 @@ object AuthenticationFilter {
   final val HTTP_CLIENT_IP_ADDRESS = new ThreadLocal[String]() {
     override protected def initialValue: String = null
   }
+  final val HTTP_PROXY_HEADER_CLIENT_IP_ADDRESS = new ThreadLocal[String]() {
+    override protected def initialValue: String = null
+  }
   final val HTTP_CLIENT_USER_NAME = new ThreadLocal[String]() {
     override protected def initialValue: String = null
   }
 
   def getUserIpAddress: String = HTTP_CLIENT_IP_ADDRESS.get
 
+  def getUserProxyHeaderIpAddress: String = HTTP_PROXY_HEADER_CLIENT_IP_ADDRESS.get()
+
   def getUserName: String = HTTP_CLIENT_USER_NAME.get
 }

kyuubi-server/src/main/scala/org/apache/kyuubi/server/http/util/SessionManager.scala

@@ -35,6 +35,20 @@ object SessionManager extends Logging {
     threadLocalIpAddress.get
   }
 
+  private val threadLocalProxyHttpHeaderIpAddress: ThreadLocal[String] = new ThreadLocal[String]
+
+  def setProxyHttpHeaderIpAddress(realIpAddress: String): Unit = {
+    threadLocalProxyHttpHeaderIpAddress.set(realIpAddress)
+  }
+
+  def clearProxyHttpHeaderIpAddress(): Unit = {
+    threadLocalProxyHttpHeaderIpAddress.remove()
+  }
+
+  def getProxyHttpHeaderIpAddress: String = {
+    threadLocalProxyHttpHeaderIpAddress.get
+  }
+
   private val threadLocalForwardedAddresses: ThreadLocal[List[String]] =
     new ThreadLocal[List[String]]