[KYUUBI #2935] Support spnego authentication for thrift http transport mode

### _Why are the changes needed?_

Reuse the AuthenticationFilter for KyuubiTHttpFrontendService

Close #2907
### _How was this patch tested?_
- [x] 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 #2935 from turboFei/http_servelet.

Closes #2935

26ba8e7 [Fei Wang] comments
241c047 [Fei Wang] refactor
7de5ffb [Fei Wang] fix
030d5ce [Fei Wang] save
f82fbb0 [Fei Wang] save
0850754 [Fei Wang] save

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

Author: turboFei

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

@@ -21,24 +21,22 @@ import java.io.IOException
 import java.security.SecureRandom
 import javax.security.sasl.AuthenticationException
 import javax.servlet.ServletException
-import javax.servlet.http.Cookie
-import javax.servlet.http.HttpServletRequest
-import javax.servlet.http.HttpServletResponse
+import javax.servlet.http.{Cookie, HttpServletRequest, HttpServletResponse}
 import javax.ws.rs.core.NewCookie
 
 import scala.collection.mutable
 
-import org.apache.commons.codec.binary.Base64
-import org.apache.commons.codec.binary.StringUtils
 import org.apache.hadoop.hive.shims.Utils
 import org.apache.thrift.TProcessor
 import org.apache.thrift.protocol.TProtocolFactory
 import org.apache.thrift.server.TServlet
 
 import org.apache.kyuubi.Logging
 import org.apache.kyuubi.config.KyuubiConf
+import org.apache.kyuubi.server.http.authentication.AuthenticationFilter
+import org.apache.kyuubi.server.http.authentication.AuthenticationHandler.AUTHORIZATION_HEADER
 import org.apache.kyuubi.server.http.util.{CookieSigner, HttpAuthUtils, SessionManager}
-import org.apache.kyuubi.service.authentication.{AuthenticationProviderFactory, KyuubiAuthenticationFactory}
+import org.apache.kyuubi.service.authentication.KyuubiAuthenticationFactory
 
 class ThriftHttpServlet(
     processor: TProcessor,
@@ -57,6 +55,7 @@ class ThriftHttpServlet(
   private var isCookieSecure = false
   private var isHttpOnlyCookie = false
   private val X_FORWARDED_FOR_HEADER = "X-Forwarded-For"
+  private val authenticationFilter = new AuthenticationFilter(conf)
 
   override def init(): Unit = {
     isCookieAuthEnabled = conf.get(KyuubiConf.FRONTEND_THRIFT_HTTP_COOKIE_AUTH_ENABLED)
@@ -72,6 +71,7 @@ class ThriftHttpServlet(
       isCookieSecure = conf.get(KyuubiConf.FRONTEND_THRIFT_HTTP_USE_SSL)
       isHttpOnlyCookie = conf.get(KyuubiConf.FRONTEND_THRIFT_HTTP_COOKIE_IS_HTTPONLY)
     }
+    authenticationFilter.initAuthHandlers()
   }
 
   @throws[ServletException]
@@ -105,10 +105,10 @@ class ThriftHttpServlet(
       // If the cookie based authentication is not enabled or the request does not have a valid
       // cookie, use authentication depending on the server setup.
       if (clientUserName == null) {
-        clientUserName = doPasswdAuth(request, authFactory)
+        clientUserName = authenticate(request, response)
       }
 
-      assert(clientUserName != null)
+      require(clientUserName != null, "No valid authorization provided")
       debug("Client username: " + clientUserName)
 
       // Set the thread local username to be used for doAs if true
@@ -280,100 +280,10 @@ class ThriftHttpServlet(
     newCookie + "; HttpOnly"
   }
 
-  /**
-   * Do the LDAP/PAM authentication
-   *
-   * @param request
-   * @param authFactory
-   * @throws AuthenticationException
-   */
-  @throws[AuthenticationException]
-  private def doPasswdAuth(
-      request: HttpServletRequest,
-      authFactory: KyuubiAuthenticationFactory): String = {
-    val userName = getUsername(request, authFactory: KyuubiAuthenticationFactory)
-    debug("Is No SASL Enabled : " + authFactory.isNoSaslEnabled)
-    // No-op when authType is NOSASL
-    if (authFactory.isNoSaslEnabled) return userName
-    try {
-      debug("Initiating Password Authentication")
-      val password = getPassword(request, authFactory)
-      val authMethod = authFactory.getValidPasswordAuthMethod
-      debug("Password Method: " + authMethod)
-      val provider = AuthenticationProviderFactory.getAuthenticationProvider(authMethod, conf)
-      debug("Password Provider obtained")
-      provider.authenticate(userName, password)
-      debug("Password Provider authenticated username successfully")
-    } catch {
-      case e: Exception =>
-        throw new AuthenticationException(e.getMessage, e)
-    }
-    userName
-  }
-
-  @throws[AuthenticationException]
-  private def getUsername(
-      request: HttpServletRequest,
-      authFactory: KyuubiAuthenticationFactory): String = {
-    val creds = getAuthHeaderTokens(request, authFactory)
-    // Username must be present
-    if (creds(0) == null || creds(0).isEmpty) {
-      throw new AuthenticationException("Authorization header received " +
-        "from the client does not contain username.")
-    }
-    creds(0)
-  }
-
-  @throws[AuthenticationException]
-  private def getPassword(
-      request: HttpServletRequest,
-      authFactory: KyuubiAuthenticationFactory): String = {
-    val creds = getAuthHeaderTokens(request, authFactory)
-    // Password must be present
-    if (creds(1) == null || creds(1).isEmpty) {
-      throw new AuthenticationException("Authorization header received " +
-        "from the client does not contain username.")
-    }
-    creds(1)
-  }
-
-  @throws[AuthenticationException]
-  private def getAuthHeaderTokens(
-      request: HttpServletRequest,
-      authFactory: KyuubiAuthenticationFactory): Array[String] = {
-    val authHeaderBase64 = getAuthHeader(request, authFactory)
-    val authHeaderString = StringUtils.newStringUtf8(Base64.decodeBase64(authHeaderBase64.getBytes))
-    authHeaderString.split(":")
-  }
-
-  /**
-   * Returns the base64 encoded auth header payload
-   *
-   * @param request
-   * @param authFactory
-   * @return
-   * @throws AuthenticationException
-   */
-  @throws[AuthenticationException]
-  private def getAuthHeader(
-      request: HttpServletRequest,
-      authFactory: KyuubiAuthenticationFactory): String = {
-    val authHeader = request.getHeader(HttpAuthUtils.AUTHORIZATION)
-    // Each http request must have an Authorization header
-    if (authHeader == null || authHeader.isEmpty) {
-      throw new AuthenticationException("Authorization header received " +
-        "from the client is empty.")
-    }
-
-    var authHeaderBase64String: String = null
-    val beginIndex = (HttpAuthUtils.BASIC + " ").length
-    authHeaderBase64String = authHeader.substring(beginIndex)
-    // Authorization header must have a payload
-    if (authHeaderBase64String == null || authHeaderBase64String.isEmpty) {
-      throw new AuthenticationException("Authorization header received " +
-        "from the client does not contain any data.")
-    }
-    authHeaderBase64String
+  private def authenticate(request: HttpServletRequest, response: HttpServletResponse): String = {
+    val authorization = request.getHeader(AUTHORIZATION_HEADER)
+    authenticationFilter.getMatchedHandler(authorization).map(
+      _.authenticate(request, response)).orNull
   }
 
   private def getDoAsQueryParam(queryString: String): String = {

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

@@ -53,7 +53,7 @@ class AuthenticationFilter(conf: KyuubiConf) extends Filter with Logging {
     }
   }
 
-  override def init(filterConfig: FilterConfig): Unit = {
+  private[kyuubi] def initAuthHandlers(): Unit = {
     val authTypes = conf.get(AUTHENTICATION_METHOD).map(AuthTypes.withName)
     val spnegoKerberosEnabled = authTypes.contains(KERBEROS)
     val basicAuthTypeOpt = {
@@ -75,9 +75,17 @@ class AuthenticationFilter(conf: KyuubiConf) extends Filter with Logging {
       val internalHandler = new KyuubiInternalAuthenticationHandler
       addAuthHandler(internalHandler)
     }
+  }
+
+  override def init(filterConfig: FilterConfig): Unit = {
+    initAuthHandlers()
     super.init(filterConfig)
   }
 
+  private[kyuubi] def getMatchedHandler(authorization: String): Option[AuthenticationHandler] = {
+    authSchemeHandlers.values.find(_.matchAuthScheme(authorization))
+  }
+
   /**
    * If the request has a valid authentication token it allows the request to continue to the
    * target resource, otherwise it triggers an authentication sequence using the configured
@@ -97,13 +105,7 @@ class AuthenticationFilter(conf: KyuubiConf) extends Filter with Logging {
     val httpResponse = response.asInstanceOf[HttpServletResponse]
 
     val authorization = httpRequest.getHeader(AUTHORIZATION_HEADER)
-    var matchedHandler: AuthenticationHandler = null
-
-    for (authHandler <- authSchemeHandlers.values if matchedHandler == null) {
-      if (authHandler.matchAuthScheme(authorization)) {
-        matchedHandler = authHandler
-      }
-    }
+    val matchedHandler = getMatchedHandler(authorization).orNull
 
     if (matchedHandler == null) {
       debug(s"No auth scheme matched for url: ${httpRequest.getRequestURL}")

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

@@ -97,7 +97,11 @@ trait AuthenticationHandler {
       throw new AuthenticationException("Authorization header received from the client is empty.")
     }
 
-    val authorization = authHeader.substring(authScheme.toString.length).trim
+    var authorization = authHeader.substring(authScheme.toString.length).trim
+    // For thrift http spnego authorization, its format is 'NEGOTIATE : $token', see HIVE-26353
+    if (authorization.startsWith(":")) {
+      authorization = authorization.stripPrefix(":").trim
+    }
     // Authorization header must have a payload
     if (authorization == null || authorization.isEmpty()) {
       throw new AuthenticationException(

kyuubi-server/src/test/scala/org/apache/kyuubi/operation/KyuubiOperationKerberosAndPlainAuthSuite.scala

@@ -32,7 +32,7 @@ class KyuubiOperationKerberosAndPlainAuthSuite extends WithKyuubiServer with Ker
   private val customPasswd: String = "password"
 
   override protected def jdbcUrl: String = getJdbcUrl
-  private def kerberosJdbcUrl: String = jdbcUrl + s"principal=${testPrincipal}"
+  protected def kerberosJdbcUrl: String = jdbcUrl.stripSuffix(";") + s";principal=${testPrincipal}"
   private val currentUser = UserGroupInformation.getCurrentUser
 
   override def beforeAll(): Unit = {

kyuubi-server/src/test/scala/org/apache/kyuubi/operation/thrift/http/KyuubiOperationThriftHttpKerberosAndPlainAuthSuite.scala

@@ -17,26 +17,43 @@
 
 package org.apache.kyuubi.operation.thrift.http
 
-import org.scalactic.source.Position
-import org.scalatest.Tag
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.security.UserGroupInformation
 
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.config.KyuubiConf.FrontendProtocols
 import org.apache.kyuubi.operation.KyuubiOperationKerberosAndPlainAuthSuite
+import org.apache.kyuubi.service.authentication.UserDefineAuthenticationProviderImpl
 
 class KyuubiOperationThriftHttpKerberosAndPlainAuthSuite
   extends KyuubiOperationKerberosAndPlainAuthSuite {
   override protected val frontendProtocols: Seq[KyuubiConf.FrontendProtocols.Value] =
     FrontendProtocols.THRIFT_HTTP :: Nil
 
+  override protected def kerberosJdbcUrl: String =
+    jdbcUrl.stripSuffix(";") + s";principal=${testSpnegoPrincipal}"
+
+  override protected lazy val conf: KyuubiConf = {
+    val config = new Configuration()
+    val authType = "hadoop.security.authentication"
+    config.set(authType, "KERBEROS")
+    System.setProperty("java.security.krb5.conf", krb5ConfPath)
+    UserGroupInformation.setConfiguration(config)
+    assert(UserGroupInformation.isSecurityEnabled)
+
+    KyuubiConf().set(KyuubiConf.AUTHENTICATION_METHOD, Seq("KERBEROS", "LDAP", "CUSTOM"))
+      .set(KyuubiConf.SERVER_KEYTAB, testKeytab)
+      .set(KyuubiConf.SERVER_PRINCIPAL, testPrincipal)
+      .set(KyuubiConf.AUTHENTICATION_LDAP_URL, ldapUrl)
+      .set(KyuubiConf.AUTHENTICATION_LDAP_BASEDN, ldapBaseDn)
+      .set(
+        KyuubiConf.AUTHENTICATION_CUSTOM_CLASS,
+        classOf[UserDefineAuthenticationProviderImpl].getCanonicalName)
+      .set(KyuubiConf.SERVER_SPNEGO_KEYTAB, testKeytab)
+      .set(KyuubiConf.SERVER_SPNEGO_PRINCIPAL, testSpnegoPrincipal)
+  }
+
   override protected def getJdbcUrl: String =
     s"jdbc:hive2://${server.frontendServices.head.connectionUrl}/default;transportMode=http;" +
       s"httpPath=cliservice"
-
-  override protected def test(testName: String, testTags: Tag*)(testFun: => Any)(implicit
-      pos: Position): Unit = {
-    if (!testName.equals("test with KERBEROS authentication")) {
-      super.test(testName, testTags: _*)(testFun)(pos)
-    }
-  }
 }