[KYUUBI #5990] Always take the first declared SASL/PLAIN auth type

# 🔍 Description
## Issue References 🔗

#5185 changed the type of `kyuubi.authentication` from `Seq`(ordered) to `Set`(unordered), which break the assumption of
```
Note that: for SASL authentication, KERBEROS and PLAIN auth types are supported
at the same time, and only the first specified PLAIN auth type is valid.
```

## Describe Your Solution 🔧

Restore the type to Seq

## Types of changes 🔖

- [x] Bugfix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to change)

## Test Plan 🧪

UT is updated

---

# Checklist 📝

- [x] This patch was not authored or co-authored using [Generative Tooling](https://www.apache.org/legal/generative-tooling.html)

**Be nice. Be informative.**

Closes #5990 from pan3793/auth-plain.

Closes #5990

acae25f [Cheng Pan] fix doc
cef7dba [Cheng Pan] fix
87b370f [Cheng Pan] Always take the first declared SASL/PLAIN auth type

Authored-by: Cheng Pan <[email protected]>
Signed-off-by: Cheng Pan <[email protected]>

docs/configuration/settings.md

@@ -33,7 +33,7 @@ You can configure the Kyuubi properties in `$KYUUBI_HOME/conf/kyuubi-defaults.co
 
 |                      Key                      |      Default      |                                                                                                                                                                                                                                                                                                                                                                                                                                                                        Meaning                                                                                                                                                                                                                                                                                                                                                                                                                                                                         |  Type  | Since |
 |-----------------------------------------------|-------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|-------|
-| kyuubi.authentication                         | NONE              | 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>JDBC: JDBC query authentication.</li> <li>LDAP: Lightweight Directory Access Protocol authentication.</li></ul>The following tree describes the catalog of each option.<ul>  <li><code>NOSASL</code></li>  <li>SASL    <ul>      <li>SASL/PLAIN</li>        <ul>          <li><code>NONE</code></li>          <li><code>LDAP</code></li>          <li><code>JDBC</code></li>          <li><code>CUSTOM</code></li>        </ul>      <li>SASL/GSSAPI        <ul>          <li><code>KERBEROS</code></li>        </ul>      </li>    </ul>  </li></ul> Note that: for SASL authentication, KERBEROS and PLAIN auth types are supported at the same time, and only the first specified PLAIN auth type is valid. | set    | 1.0.0 |
+| kyuubi.authentication                         | NONE              | 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>JDBC: JDBC query authentication.</li> <li>LDAP: Lightweight Directory Access Protocol authentication.</li></ul>The following tree describes the catalog of each option.<ul>  <li><code>NOSASL</code></li>  <li>SASL    <ul>      <li>SASL/PLAIN</li>        <ul>          <li><code>NONE</code></li>          <li><code>LDAP</code></li>          <li><code>JDBC</code></li>          <li><code>CUSTOM</code></li>        </ul>      <li>SASL/GSSAPI        <ul>          <li><code>KERBEROS</code></li>        </ul>      </li>    </ul>  </li></ul> Note that: for SASL authentication, KERBEROS and PLAIN auth types are supported at the same time, and only the first specified PLAIN auth type is valid. | seq    | 1.0.0 |
 | kyuubi.authentication.custom.class            | &lt;undefined&gt; | User-defined authentication implementation of org.apache.kyuubi.service.authentication.PasswdAuthenticationProvider                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    | string | 1.3.0 |
 | kyuubi.authentication.jdbc.driver.class       | &lt;undefined&gt; | Driver class name for JDBC Authentication Provider.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    | string | 1.6.0 |
 | kyuubi.authentication.jdbc.password           | &lt;undefined&gt; | Database password for JDBC Authentication Provider.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    | string | 1.6.0 |

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

@@ -757,7 +757,7 @@ object KyuubiConf {
       .stringConf
       .createWithDefault("X-Real-IP")
 
-  val AUTHENTICATION_METHOD: ConfigEntry[Set[String]] = buildConf("kyuubi.authentication")
+  val AUTHENTICATION_METHOD: ConfigEntry[Seq[String]] = buildConf("kyuubi.authentication")
     .doc("A comma-separated list of client authentication types." +
       "<ul>" +
       " <li>NOSASL: raw transport.</li>" +
@@ -793,9 +793,9 @@ object KyuubiConf {
     .serverOnly
     .stringConf
     .transformToUpperCase
-    .toSet()
+    .toSequence()
     .checkValues(AuthTypes)
-    .createWithDefault(Set(AuthTypes.NONE.toString))
+    .createWithDefault(Seq(AuthTypes.NONE.toString))
 
   val AUTHENTICATION_CUSTOM_CLASS: OptionalConfigEntry[String] =
     buildConf("kyuubi.authentication.custom.class")

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

@@ -37,11 +37,15 @@ import org.apache.kyuubi.shaded.thrift.transport.{TSaslServerTransport, TTranspo
 
 class KyuubiAuthenticationFactory(conf: KyuubiConf, isServer: Boolean = true) extends Logging {
 
-  val authTypes: Set[AuthType] = conf.get(AUTHENTICATION_METHOD).map(AuthTypes.withName)
-  val noSaslEnabled: Boolean = authTypes == Set(NOSASL)
+  val authTypes: Seq[AuthType] = conf.get(AUTHENTICATION_METHOD).map(AuthTypes.withName)
+  val saslDisabled: Boolean = authTypes == Seq(NOSASL)
   val kerberosEnabled: Boolean = authTypes.contains(KERBEROS)
-  private val plainAuthTypeOpt = authTypes.filterNot(_.equals(KERBEROS))
-    .filterNot(_.equals(NOSASL)).headOption
+
+  // take the first declared SASL/PLAIN auth type
+  private val effectivePlainAuthType = authTypes.find {
+    case NOSASL | KERBEROS => false
+    case _ => true
+  }
 
   private val hadoopAuthServer: Option[HadoopThriftAuthBridgeServer] = {
     if (kerberosEnabled) {
@@ -70,7 +74,7 @@ class KyuubiAuthenticationFactory(conf: KyuubiConf, isServer: Boolean = true) ex
   }
 
   def getTTransportFactory: TTransportFactory = {
-    if (noSaslEnabled) {
+    if (saslDisabled) {
       new TTransportFactory()
     } else {
       var transportFactory: TSaslServerTransport.Factory = null
@@ -87,7 +91,7 @@ class KyuubiAuthenticationFactory(conf: KyuubiConf, isServer: Boolean = true) ex
         case _ =>
       }
 
-      plainAuthTypeOpt match {
+      effectivePlainAuthType match {
         case Some(plainAuthType) =>
           transportFactory = PlainSASLHelper.getTransportFactory(
             plainAuthType.toString,
@@ -152,8 +156,8 @@ object KyuubiAuthenticationFactory extends Logging {
     }
   }
 
-  def getValidPasswordAuthMethod(authTypes: Set[AuthType]): AuthMethod = {
-    if (authTypes == Set(NOSASL)) AuthMethods.NONE
+  def getValidPasswordAuthMethod(authTypes: Seq[AuthType]): AuthMethod = {
+    if (authTypes == Seq(NOSASL)) AuthMethods.NONE
     else if (authTypes.contains(NONE)) AuthMethods.NONE
     else if (authTypes.contains(LDAP)) AuthMethods.LDAP
     else if (authTypes.contains(JDBC)) AuthMethods.JDBC

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

@@ -56,33 +56,33 @@ class KyuubiAuthenticationFactorySuite extends KyuubiFunSuite {
   }
 
   test("AuthType Other") {
-    val conf = KyuubiConf().set(KyuubiConf.AUTHENTICATION_METHOD, Set("INVALID"))
+    val conf = KyuubiConf().set(KyuubiConf.AUTHENTICATION_METHOD, Seq("INVALID"))
     interceptEquals[IllegalArgumentException] { new KyuubiAuthenticationFactory(conf) }(
       "The value of kyuubi.authentication should be one of" +
         " NOSASL, NONE, LDAP, JDBC, KERBEROS, CUSTOM, but was INVALID")
   }
 
   test("AuthType LDAP") {
-    val conf = KyuubiConf().set(KyuubiConf.AUTHENTICATION_METHOD, Set("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, Set("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, Set("NOSASL"))
+    val conf = KyuubiConf().set(KyuubiConf.AUTHENTICATION_METHOD, Seq("NOSASL"))
     var factory = new KyuubiAuthenticationFactory(conf)
     !factory.getTTransportFactory.isInstanceOf[TSaslServerTransport.Factory]
 
-    conf.set(KyuubiConf.AUTHENTICATION_METHOD, Set("NOSASL", "NONE"))
+    conf.set(KyuubiConf.AUTHENTICATION_METHOD, Seq("NOSASL", "NONE"))
     factory = new KyuubiAuthenticationFactory(conf)
     factory.getTTransportFactory.isInstanceOf[TSaslServerTransport.Factory]
   }

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

@@ -136,7 +136,7 @@ class ThriftHttpServlet(
       } else SessionManager.setForwardedAddresses(List.empty[String])
 
       // Generate new cookie and add it to the response
-      if (requireNewCookie && !authFactory.noSaslEnabled) {
+      if (requireNewCookie && !authFactory.saslDisabled) {
         val cookieToken = HttpAuthUtils.createCookieToken(clientUserName)
         val hs2Cookie = createCookie(signer.signCookie(cookieToken))
         if (isHttpOnlyCookie) response.setHeader("SET-COOKIE", getHttpOnlyCookieHeader(hs2Cookie))

kyuubi-server/src/test/scala/org/apache/kyuubi/RestClientTestHelper.scala

@@ -48,7 +48,7 @@ trait RestClientTestHelper extends RestFrontendTestHelper with KerberizedTestHel
     UserGroupInformation.setConfiguration(config)
     assert(UserGroupInformation.isSecurityEnabled)
 
-    val conf = KyuubiConf().set(KyuubiConf.AUTHENTICATION_METHOD, Set("KERBEROS", "LDAP", "CUSTOM"))
+    val conf = KyuubiConf().set(KyuubiConf.AUTHENTICATION_METHOD, Seq("KERBEROS", "LDAP", "CUSTOM"))
       .set(KyuubiConf.SERVER_KEYTAB.key, testKeytab)
       .set(KyuubiConf.SERVER_PRINCIPAL, testPrincipal)
       .set(KyuubiConf.SERVER_SPNEGO_KEYTAB, testKeytab)

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

@@ -64,7 +64,7 @@ class KyuubiOperationKerberosAndPlainAuthSuite extends WithKyuubiServer with Ker
     assert(UserGroupInformation.isSecurityEnabled)
 
     KyuubiConf()
-      .set(KyuubiConf.AUTHENTICATION_METHOD, Set("KERBEROS", "LDAP", "CUSTOM"))
+      .set(KyuubiConf.AUTHENTICATION_METHOD, Seq("KERBEROS", "LDAP", "CUSTOM"))
       .set(KyuubiConf.SERVER_KEYTAB, testKeytab)
       .set(KyuubiConf.SERVER_PRINCIPAL, testPrincipal)
       .set(KyuubiConf.AUTHENTICATION_LDAP_URL, ldapUrl)

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

@@ -49,7 +49,7 @@ class KyuubiOperationThriftHttpKerberosAndPlainAuthSuite
     UserGroupInformation.setConfiguration(config)
     assert(UserGroupInformation.isSecurityEnabled)
 
-    KyuubiConf().set(KyuubiConf.AUTHENTICATION_METHOD, Set("KERBEROS", "LDAP", "CUSTOM"))
+    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)

kyuubi-server/src/test/scala/org/apache/kyuubi/server/api/v1/AdminResourceSuite.scala

@@ -51,7 +51,7 @@ class AdminResourceSuite extends KyuubiFunSuite with RestFrontendTestHelper {
   private val engineMgr = new KyuubiApplicationManager()
 
   override protected lazy val conf: KyuubiConf = KyuubiConf()
-    .set(AUTHENTICATION_METHOD, Set("CUSTOM"))
+    .set(AUTHENTICATION_METHOD, Seq("CUSTOM"))
     .set(AUTHENTICATION_CUSTOM_CLASS, classOf[AnonymousAuthenticationProviderImpl].getName)
     .set(SERVER_ADMINISTRATORS, Set("admin001"))
     .set(ENGINE_IDLE_TIMEOUT, Duration.ofMinutes(3).toMillis)

kyuubi-server/src/test/scala/org/apache/kyuubi/server/api/v1/BatchesResourceSuite.scala

@@ -85,7 +85,7 @@ abstract class BatchesResourceSuiteBase extends KyuubiFunSuite
   override protected lazy val conf: KyuubiConf = {
     val testResourceDir = Paths.get(sparkBatchTestResource.get).getParent
     val kyuubiConf = KyuubiConf()
-      .set(AUTHENTICATION_METHOD, Set("CUSTOM"))
+      .set(AUTHENTICATION_METHOD, Seq("CUSTOM"))
       .set(AUTHENTICATION_CUSTOM_CLASS, classOf[AnonymousAuthenticationProviderImpl].getName)
       .set(SERVER_ADMINISTRATORS, Set("admin"))
       .set(BATCH_IMPL_VERSION, batchVersion)

kyuubi-server/src/test/scala/org/apache/kyuubi/server/rest/client/AdminCtlSuite.scala

@@ -56,7 +56,7 @@ class AdminCtlSuite extends RestClientTestHelper with TestPrematureExit {
     val id = UUID.randomUUID().toString
     conf.set(HighAvailabilityConf.HA_NAMESPACE, "kyuubi_test")
     conf.set(KyuubiConf.ENGINE_IDLE_TIMEOUT, 180000L)
-    conf.set(KyuubiConf.AUTHENTICATION_METHOD, Set("LDAP", "CUSTOM"))
+    conf.set(KyuubiConf.AUTHENTICATION_METHOD, Seq("LDAP", "CUSTOM"))
     conf.set(KyuubiConf.GROUP_PROVIDER, "hadoop")
 
     val user = ldapUser

kyuubi-server/src/test/scala/org/apache/kyuubi/server/rest/client/AdminRestApiSuite.scala

@@ -50,7 +50,7 @@ class AdminRestApiSuite extends RestClientTestHelper {
     val id = UUID.randomUUID().toString
     conf.set(HighAvailabilityConf.HA_NAMESPACE, "kyuubi_test")
     conf.set(KyuubiConf.ENGINE_IDLE_TIMEOUT, 180000L)
-    conf.set(KyuubiConf.AUTHENTICATION_METHOD, Set("LDAP", "CUSTOM"))
+    conf.set(KyuubiConf.AUTHENTICATION_METHOD, Seq("LDAP", "CUSTOM"))
     conf.set(KyuubiConf.GROUP_PROVIDER, "hadoop")
     val user = ldapUser
     val engine = new EngineRef(conf.clone, user, PluginLoader.loadGroupProvider(conf), id, null)