[KYUUBI #1262] Support both KERBEROS and PLAIN authentication at the same time

docs/deployment/settings.md

@@ -122,7 +122,7 @@ You can configure the Kyuubi properties in `$KYUUBI_HOME/conf/kyuubi-defaults.co
 
 Key | Default | Meaning | Type | Since
 --- | --- | --- | --- | ---
-kyuubi\.authentication|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>NONE</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>Client authentication types.<ul> <li>NOSASL: raw transport.</li> <li>NONE: no authentication check.</li> <li>KERBEROS: Kerberos/GSSAPI authentication.</li> <li>CUSTOM: User-defined authentication.</li> <li>LDAP: Lightweight Directory Access Protocol authentication.</li></ul></div>|<div style='width: 30pt'>string</div>|<div style='width: 20pt'>1.0.0</div>
+kyuubi\.authentication|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>NONE</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>A comma separated list of client authentication types.<ul> <li>NOSASL: raw transport.</li> <li>NONE: no authentication check.</li> <li>KERBEROS: Kerberos/GSSAPI authentication.</li> <li>CUSTOM: User-defined authentication.</li> <li>LDAP: Lightweight Directory Access Protocol authentication.</li></ul> Note that: For KERBEROS, it is SASL/GSSAPI mechanism, and for NONE, CUSTOM and LDAP, they are all SASL/PLAIN mechanism. If only NOSASL is specified, the authentication will be NOSASL. For SASL authentication, KERBEROS and PLAIN auth type are supported at the same time, and only the first specified PLAIN auth type is valid.</div>|<div style='width: 30pt'>seq</div>|<div style='width: 20pt'>1.0.0</div>
 kyuubi\.authentication<br>\.custom\.class|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>&lt;undefined&gt;</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>User-defined authentication implementation of org.apache.kyuubi.service.authentication.PasswdAuthenticationProvider</div>|<div style='width: 30pt'>string</div>|<div style='width: 20pt'>1.3.0</div>
 kyuubi\.authentication<br>\.ldap\.base\.dn|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>&lt;undefined&gt;</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>LDAP base DN.</div>|<div style='width: 30pt'>string</div>|<div style='width: 20pt'>1.0.0</div>
 kyuubi\.authentication<br>\.ldap\.domain|<div style='width: 65pt;word-wrap: break-word;white-space: normal'>&lt;undefined&gt;</div>|<div style='width: 170pt;word-wrap: break-word;white-space: normal'>LDAP domain.</div>|<div style='width: 30pt'>string</div>|<div style='width: 20pt'>1.0.0</div>

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

@@ -372,18 +372,25 @@ object KyuubiConf {
       .version("1.4.0")
       .fallbackConf(FRONTEND_LOGIN_BACKOFF_SLOT_LENGTH)
 
-  val AUTHENTICATION_METHOD: ConfigEntry[String] = buildConf("authentication")
-    .doc("Client authentication types.<ul>" +
+  val AUTHENTICATION_METHOD: ConfigEntry[Seq[String]] = buildConf("authentication")
+    .doc("A comma separated list of client authentication types.<ul>" +
       " <li>NOSASL: raw transport.</li>" +
       " <li>NONE: no authentication check.</li>" +
       " <li>KERBEROS: Kerberos/GSSAPI authentication.</li>" +
       " <li>CUSTOM: User-defined authentication.</li>" +
-      " <li>LDAP: Lightweight Directory Access Protocol authentication.</li></ul>")
+      " <li>LDAP: Lightweight Directory Access Protocol authentication.</li></ul>" +
+      " Note that: For KERBEROS, it is SASL/GSSAPI mechanism," +
+      " and for NONE, CUSTOM and LDAP, they are all SASL/PLAIN mechanism." +
+      " If only NOSASL is specified, the authentication will be NOSASL." +
+      " For SASL authentication, KERBEROS and PLAIN auth type are supported at the same time," +
+      " and only the first specified PLAIN auth type is valid.")
     .version("1.0.0")
     .stringConf
-    .transform(_.toUpperCase(Locale.ROOT))
-    .checkValues(AuthTypes.values.map(_.toString))
-    .createWithDefault(AuthTypes.NONE.toString)
+    .toSequence()
+    .transform(_.map(_.toUpperCase(Locale.ROOT)))
+    .checkValue(_.forall(AuthTypes.values.map(_.toString).contains),
+      s"the authentication type should be one or more of ${AuthTypes.values.mkString(",")}")
+    .createWithDefault(Seq(AuthTypes.NONE.toString))
 
   val AUTHENTICATION_CUSTOM_CLASS: OptionalConfigEntry[String] =
     buildConf("authentication.custom.class")

kyuubi-common/src/main/scala/org/apache/kyuubi/service/authentication/KyuubiAuthenticationFactory.scala

@@ -27,27 +27,33 @@ import org.apache.hadoop.security.authentication.util.KerberosName
 import org.apache.hadoop.security.authorize.ProxyUsers
 import org.apache.hive.service.rpc.thrift.TCLIService.Iface
 import org.apache.thrift.TProcessorFactory
-import org.apache.thrift.transport.{TTransportException, TTransportFactory}
+import org.apache.thrift.transport.{TSaslServerTransport, TTransportException, TTransportFactory}
 
-import org.apache.kyuubi.KyuubiSQLException
+import org.apache.kyuubi.{KyuubiSQLException, Logging}
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.config.KyuubiConf._
 import org.apache.kyuubi.service.authentication.AuthTypes._
 
-class KyuubiAuthenticationFactory(conf: KyuubiConf) {
+class KyuubiAuthenticationFactory(conf: KyuubiConf) extends Logging {
 
-  private val authType: AuthType = AuthTypes.withName(conf.get(AUTHENTICATION_METHOD))
+  private val authTypes = conf.get(AUTHENTICATION_METHOD).map(AuthTypes.withName)
+  private val noSasl = authTypes == Seq(NOSASL)
+  private val kerberosEnabled = authTypes.contains(KERBEROS)
+  private val plainAuthTypeOpt = authTypes.filterNot(_.equals(KERBEROS))
+    .filterNot(_.equals(NOSASL)).headOption
 
-  private val saslServer: Option[HadoopThriftAuthBridgeServer] = authType match {
-    case KERBEROS =>
+  private val hadoopAuthServer: Option[HadoopThriftAuthBridgeServer] = {
+    if (kerberosEnabled) {
       val secretMgr = KyuubiDelegationTokenManager(conf)
       try {
         secretMgr.startThreads()
       } catch {
         case e: IOException => throw new TTransportException("Failed to start token manager", e)
       }
       Some(new HadoopThriftAuthBridgeServer(secretMgr))
-    case _ => None
+    } else {
+      None
+    }
   }
 
   private def getSaslProperties: java.util.Map[String, String] = {
@@ -59,34 +65,48 @@ class KyuubiAuthenticationFactory(conf: KyuubiConf) {
   }
 
   def getTTransportFactory: TTransportFactory = {
-    saslServer match {
-      case Some(server) =>
-        val serverTransportFactory = try {
-          server.createSaslServerTransportFactory(getSaslProperties)
-        } catch {
-          case e: TTransportException => throw new LoginException(e.getMessage)
-        }
+    if (noSasl) {
+      new TTransportFactory()
+    } else {
+      var transportFactory: TSaslServerTransport.Factory = null
+
+      hadoopAuthServer match {
+        case Some(server) =>
+          transportFactory = try {
+            server.createSaslServerTransportFactory(getSaslProperties)
+          } catch {
+            case e: TTransportException => throw new LoginException(e.getMessage)
+          }
+
+        case _ =>
+      }
 
-        server.wrapTransportFactory(serverTransportFactory)
+      plainAuthTypeOpt match {
+        case Some(plainAuthType) =>
+          transportFactory = PlainSASLHelper.getTransportFactory(plainAuthType.toString, conf,
+            Option(transportFactory)).asInstanceOf[TSaslServerTransport.Factory]
+
+        case _ =>
+      }
 
-      case _ => authType match {
-        case NOSASL => new TTransportFactory
-        case _ => PlainSASLHelper.getTransportFactory(authType.toString, conf)
+      hadoopAuthServer match {
+        case Some(server) => server.wrapTransportFactory(transportFactory)
+        case _ => transportFactory
       }
     }
   }
 
-  def getTProcessorFactory(fe: Iface): TProcessorFactory = saslServer match {
+  def getTProcessorFactory(fe: Iface): TProcessorFactory = hadoopAuthServer match {
     case Some(server) => FEServiceProcessorFactory(server, fe)
     case _ => PlainSASLHelper.getProcessFactory(fe)
   }
 
   def getRemoteUser: Option[String] = {
-    saslServer.map(_.getRemoteUser).orElse(Option(TSetIpAddressProcessor.getUserName))
+    hadoopAuthServer.map(_.getRemoteUser).orElse(Option(TSetIpAddressProcessor.getUserName))
   }
 
   def getIpAddress: Option[String] = {
-    saslServer.map(_.getRemoteAddress).map(_.getHostAddress)
+    hadoopAuthServer.map(_.getRemoteAddress).map(_.getHostAddress)
       .orElse(Option(TSetIpAddressProcessor.getUserIpAddress))
   }
 }

kyuubi-common/src/main/scala/org/apache/kyuubi/service/authentication/PlainSASLHelper.scala

@@ -71,8 +71,11 @@ object PlainSASLHelper {
     SQLPlainProcessorFactory(service)
   }
 
-  def getTransportFactory(authTypeStr: String, conf: KyuubiConf): TTransportFactory = {
-    val saslFactory = new TSaslServerTransport.Factory()
+  def getTransportFactory(
+      authTypeStr: String,
+      conf: KyuubiConf,
+      transportFactory: Option[TSaslServerTransport.Factory] = None): TTransportFactory = {
+    val saslFactory = transportFactory.getOrElse(new TSaslServerTransport.Factory())
     try {
       val handler = new PlainServerCallbackHandler(authTypeStr, conf)
       val props = new java.util.HashMap[String, String]

kyuubi-common/src/test/scala/org/apache/kyuubi/KerberizedTestHelper.scala

@@ -48,28 +48,35 @@ trait KerberizedTestHelper extends KyuubiFunSuite {
   kdcConf.remove(MiniKdc.DEBUG)
 
   private var kdc: MiniKdc = _
-  private var krb5ConfPath: String = _
-
-  eventually(timeout(60.seconds), interval(1.second)) {
-    try {
-      kdc = new MiniKdc(kdcConf, baseDir)
-      kdc.start()
-      krb5ConfPath = kdc.getKrb5conf.getAbsolutePath
-    } catch {
-      case NonFatal(e) =>
-        if (kdc != null) {
-          kdc.stop()
-          kdc = null
-        }
-        throw e
-    }
-  }
+  protected var krb5ConfPath: String = _
 
   private val keytabFile = new File(baseDir, "kyuubi-test.keytab")
   protected val testKeytab: String = keytabFile.getAbsolutePath
-  protected var testPrincipal = s"client/$hostName"
-  kdc.createPrincipal(keytabFile, testPrincipal)
-
+  protected var testPrincipal: String = _
+
+  override def beforeAll(): Unit = {
+    eventually(timeout(60.seconds), interval(1.second)) {
+      try {
+        kdc = new MiniKdc(kdcConf, baseDir)
+        kdc.start()
+        krb5ConfPath = kdc.getKrb5conf.getAbsolutePath
+      } catch {
+        case NonFatal(e) =>
+          if (kdc != null) {
+            kdc.stop()
+            kdc = null
+          }
+          throw e
+      }
+    }
+    val tempTestPrincipal = s"client/$hostName"
+    kdc.createPrincipal(keytabFile, tempTestPrincipal)
+    rewriteKrb5Conf()
+    testPrincipal = tempTestPrincipal + "@" + kdc.getRealm
+    info(s"KerberizedTest Principal: $testPrincipal")
+    info(s"KerberizedTest Keytab: $testKeytab")
+    super.beforeAll()
+  }
 
   /**
    * Forked from Apache Spark
@@ -113,13 +120,6 @@ trait KerberizedTestHelper extends KyuubiFunSuite {
     System.lineSeparator() + s"    $key=$value"
   }
 
-  rewriteKrb5Conf()
-
-  testPrincipal = testPrincipal + "@" + kdc.getRealm
-
-  info(s"KerberizedTest Principal: $testPrincipal")
-  info(s"KerberizedTest Keytab: $testKeytab")
-
   override def afterAll(): Unit = {
     kdc.stop()
     super.afterAll()

kyuubi-common/src/test/scala/org/apache/kyuubi/operation/JDBCTestUtils.scala

@@ -63,7 +63,9 @@ trait JDBCTestUtils extends KyuubiFunSuite {
     }
   }
 
-  private def jdbcUrlWithConf: String = {
+  protected def jdbcUrlWithConf: String = jdbcUrlWithConf(jdbcUrl)
+
+  protected def jdbcUrlWithConf(jdbcUrl: String): String = {
     val sessionConfStr = sessionConfigs.map(kv => kv._1 + "=" + kv._2).mkString(";")
     val sparkHiveConfStr = if (sparkHiveConfigs.isEmpty) {
       ""

kyuubi-common/src/test/scala/org/apache/kyuubi/service/authentication/CustomAuthenticationProviderImplSuite.scala

@@ -33,7 +33,7 @@ class CustomAuthenticationProviderImplSuite extends KyuubiFunSuite {
       "authentication.custom.class must be set when auth method was CUSTOM."))
 
     conf.set(KyuubiConf.AUTHENTICATION_CUSTOM_CLASS,
-      "org.apache.kyuubi.service.authentication.UserDefineAuthenticationProviderImpl")
+      classOf[UserDefineAuthenticationProviderImpl].getCanonicalName)
     val p1 = getAuthenticationProvider(AuthMethods.withName("CUSTOM"), conf)
     val e2 = intercept[AuthenticationException](p1.authenticate("test", "test"))
     assert(e2.getMessage.contains("Username or password is not valid!"))

kyuubi-common/src/test/scala/org/apache/kyuubi/service/authentication/KyuubiAuthenticationFactorySuite.scala

@@ -20,12 +20,13 @@ package org.apache.kyuubi.service.authentication
 import java.security.Security
 import javax.security.auth.login.LoginException
 
+import org.apache.thrift.transport.TSaslServerTransport
+
 import org.apache.kyuubi.{KyuubiFunSuite, KyuubiSQLException}
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.service.authentication.PlainSASLServer.SaslPlainProvider
 import org.apache.kyuubi.util.KyuubiHadoopUtils
 
-
 class KyuubiAuthenticationFactorySuite extends KyuubiFunSuite {
   import KyuubiAuthenticationFactory._
 
@@ -55,25 +56,35 @@ class KyuubiAuthenticationFactorySuite extends KyuubiFunSuite {
   }
 
   test("AuthType Other") {
-    val conf = KyuubiConf().set(KyuubiConf.AUTHENTICATION_METHOD, "INVALID")
+    val conf = KyuubiConf().set(KyuubiConf.AUTHENTICATION_METHOD, Seq("INVALID"))
     val e = intercept[IllegalArgumentException](new KyuubiAuthenticationFactory(conf))
-    assert(e.getMessage === "The value of kyuubi.authentication should be one of" +
-      " CUSTOM, KERBEROS, LDAP, NONE, NOSASL, but was INVALID")
+    assert(e.getMessage === "the authentication type should be one or more of" +
+      " NOSASL,NONE,LDAP,KERBEROS,CUSTOM")
   }
 
   test("AuthType LDAP") {
-    val conf = KyuubiConf().set(KyuubiConf.AUTHENTICATION_METHOD, "LDAP")
+    val conf = KyuubiConf().set(KyuubiConf.AUTHENTICATION_METHOD, Seq("LDAP"))
     val authFactory = new KyuubiAuthenticationFactory(conf)
     authFactory.getTTransportFactory
     assert(Security.getProviders.exists(_.isInstanceOf[SaslPlainProvider]))
   }
 
 
   test("AuthType KERBEROS w/o keytab/principal") {
-    val conf = KyuubiConf().set(KyuubiConf.AUTHENTICATION_METHOD, "KERBEROS")
+    val conf = KyuubiConf().set(KyuubiConf.AUTHENTICATION_METHOD, Seq("KERBEROS"))
 
     val factory = new KyuubiAuthenticationFactory(conf)
     val e = intercept[LoginException](factory.getTTransportFactory)
     assert(e.getMessage startsWith "Kerberos principal should have 3 parts")
   }
+
+  test("AuthType is NOSASL if only NOSASL is specified") {
+    val conf = KyuubiConf().set(KyuubiConf.AUTHENTICATION_METHOD, Seq("NOSASL"))
+    var factory = new KyuubiAuthenticationFactory(conf)
+    !factory.getTTransportFactory.isInstanceOf[TSaslServerTransport.Factory]
+
+    conf.set(KyuubiConf.AUTHENTICATION_METHOD, Seq("NOSASL", "NONE"))
+    factory = new KyuubiAuthenticationFactory(conf)
+    factory.getTTransportFactory.isInstanceOf[TSaslServerTransport.Factory]
+  }
 }

kyuubi-common/src/test/scala/org/apache/kyuubi/service/authentication/LdapAuthenticationProviderImplSuite.scala

@@ -20,28 +20,18 @@ package org.apache.kyuubi.service.authentication
 import javax.naming.CommunicationException
 import javax.security.sasl.AuthenticationException
 
-import com.unboundid.ldap.listener.InMemoryDirectoryServer
-import com.unboundid.ldap.listener.InMemoryDirectoryServerConfig
-
-import org.apache.kyuubi.KyuubiFunSuite
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.config.KyuubiConf._
 
-class LdapAuthenticationProviderImplSuite extends KyuubiFunSuite {
-
-  private var ldapServer: InMemoryDirectoryServer = _
+class LdapAuthenticationProviderImplSuite extends WithLdapServer {
+  override protected val ldapUser: String = "kentyao"
+  override protected val ldapUserPasswd: String = "kentyao"
 
   private val conf = new KyuubiConf()
 
   override def beforeAll(): Unit = {
-    val config = new InMemoryDirectoryServerConfig("ou=users")
-    config.addAdditionalBindCredentials("uid=kentyao,ou=users", "kentyao")
-    ldapServer = new InMemoryDirectoryServer(config)
-    ldapServer.startListening()
-
-    conf.set(AUTHENTICATION_LDAP_URL, s"ldap://localhost:" + ldapServer.getListenPort)
-
     super.beforeAll()
+    conf.set(AUTHENTICATION_LDAP_URL, ldapUrl)
   }
 
   override def afterAll(): Unit = {
@@ -67,8 +57,7 @@ class LdapAuthenticationProviderImplSuite extends KyuubiFunSuite {
     assert(e3.getMessage.contains(user))
     assert(e3.getCause.isInstanceOf[javax.naming.AuthenticationException])
 
-    val dn = "ou=users"
-    conf.set(AUTHENTICATION_LDAP_BASEDN, dn)
+    conf.set(AUTHENTICATION_LDAP_BASEDN, ldapBaseDn)
     val providerImpl2 = new LdapAuthenticationProviderImpl(conf)
     providerImpl2.authenticate("kentyao", "kentyao")
 

kyuubi-common/src/test/scala/org/apache/kyuubi/service/authentication/WithLdapServer.scala

@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.service.authentication
+
+import com.unboundid.ldap.listener.{InMemoryDirectoryServer, InMemoryDirectoryServerConfig}
+
+import org.apache.kyuubi.{KyuubiFunSuite, Utils}
+
+trait WithLdapServer extends KyuubiFunSuite {
+  protected var ldapServer: InMemoryDirectoryServer = _
+  protected val ldapBaseDn = "ou=users"
+  protected val ldapUser = Utils.currentUser
+  protected val ldapUserPasswd = "ldapPassword"
+
+  protected def ldapUrl = s"ldap://localhost:${ldapServer.getListenPort}"
+
+  override def beforeAll(): Unit = {
+    val config = new InMemoryDirectoryServerConfig(ldapBaseDn)
+    config.addAdditionalBindCredentials(s"uid=$ldapUser,ou=users", ldapUserPasswd)
+    ldapServer = new InMemoryDirectoryServer(config)
+    ldapServer.startListening()
+    super.beforeAll()
+  }
+
+  override def afterAll(): Unit = {
+    ldapServer.close()
+    super.afterAll()
+  }
+}