[KYUUBI #6041] RESTful API supports isolated authentication enable co…

…nfiguration

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

@@ -797,6 +797,14 @@ object KyuubiConf {
     .checkValues(AuthTypes)
     .createWithDefault(Seq(AuthTypes.NONE.toString))
 
+  val RESTFUL_AUTHENTICATION: ConfigEntry[Boolean] =
+    buildConf("kyuubi.restful.security.enabled")
+      .doc("Whether to enable kyuubi server restful api secure access.")
+      .version("1.9.0")
+      .serverOnly
+      .booleanConf
+      .createWithDefault(true)
+
   val AUTHENTICATION_CUSTOM_CLASS: OptionalConfigEntry[String] =
     buildConf("kyuubi.authentication.custom.class")
       .doc("User-defined authentication implementation of " +

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

@@ -26,7 +26,7 @@ import scala.collection.mutable
 
 import org.apache.kyuubi.Logging
 import org.apache.kyuubi.config.KyuubiConf
-import org.apache.kyuubi.config.KyuubiConf.{AUTHENTICATION_METHOD, FRONTEND_PROXY_HTTP_CLIENT_IP_HEADER}
+import org.apache.kyuubi.config.KyuubiConf.{AUTHENTICATION_METHOD, FRONTEND_PROXY_HTTP_CLIENT_IP_HEADER, RESTFUL_AUTHENTICATION}
 import org.apache.kyuubi.server.http.util.HttpAuthUtils.AUTHORIZATION_HEADER
 import org.apache.kyuubi.service.authentication.{AuthTypes, InternalSecurityAccessor}
 import org.apache.kyuubi.service.authentication.AuthTypes.{KERBEROS, NOSASL}
@@ -79,7 +79,14 @@ class AuthenticationFilter(conf: KyuubiConf) extends Filter with Logging {
   }
 
   override def init(filterConfig: FilterConfig): Unit = {
-    initAuthHandlers()
+    if (conf.get(RESTFUL_AUTHENTICATION)) {
+      initAuthHandlers()
+    } else {
+      Set(NOSASL).foreach { basicAuthType =>
+        val basicHandler = new BasicAuthenticationHandler(basicAuthType)
+        addAuthHandler(basicHandler)
+      }
+    }
   }
 
   private[kyuubi] def getMatchedHandler(authorization: String): Option[AuthenticationHandler] = {

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

@@ -0,0 +1,50 @@
+/*
+ * 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.operation
+
+import javax.servlet.http.HttpServletResponse
+
+import org.apache.kyuubi.config.KyuubiConf
+import org.apache.kyuubi.service.authentication.InternalSecurityAccessor
+
+class KyuubiRestDisableAuthenticationSuite extends KyuubiRestAuthenticationSuite {
+
+  override protected val otherConfigs: Map[String, String] = {
+    Map(
+      KyuubiConf.RESTFUL_AUTHENTICATION.key -> "false",
+      KyuubiConf.ENGINE_SECURITY_ENABLED.key -> "true",
+      KyuubiConf.ENGINE_SECURITY_SECRET_PROVIDER.key -> "simple",
+      KyuubiConf.SIMPLE_SECURITY_SECRET_PROVIDER_PROVIDER_SECRET.key -> "_KYUUBI_REST_",
+      // allow to impersonate other users with spnego authentication
+      s"hadoop.proxyuser.$clientPrincipalUser.groups" -> "*",
+      s"hadoop.proxyuser.$clientPrincipalUser.hosts" -> "*")
+  }
+
+  override def beforeAll(): Unit = {
+    super.beforeAll()
+    InternalSecurityAccessor.initialize(conf, true)
+  }
+
+  test("test without authorization when disable kyuubi.restful.security.enabled") {
+    val response = webTarget.path("api/v1/sessions/count")
+      .request()
+      .get()
+
+    assert(HttpServletResponse.SC_OK == response.getStatus)
+  }
+}