feat(ranger): added delegation token support (v1.10.1).

extensions/spark/kyuubi-spark-authz/pom.xml

@@ -37,7 +37,7 @@
         <gethostname4j.version>1.0.0</gethostname4j.version>
         <jersey.client.version>2.40</jersey.client.version>
         <jna.version>5.7.0</jna.version>
-        <solr.version>8.11.3</solr.version>
+        <solr.version>8.11.4</solr.version>
     </properties>
 
     <dependencies>
@@ -396,6 +396,17 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.solr</groupId>
+            <artifactId>solr-hadoop-auth-framework</artifactId>
+            <version>${solr.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.hadoop</groupId>
+                    <artifactId>*</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
         <dependency>
             <groupId>org.apache.solr</groupId>
             <artifactId>solr-solrj</artifactId>

extensions/spark/kyuubi-spark-authz/src/main/resources/META-INF/services/org.apache.spark.security.HadoopDelegationTokenProvider

@@ -0,0 +1,19 @@
+#
+# 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.
+#
+
+org.apache.kyuubi.plugin.spark.authz.ranger.RangerDelegationTokenProvider
+org.apache.kyuubi.plugin.spark.authz.ranger.SolrDelegationTokenProvider

extensions/spark/kyuubi-spark-authz/src/main/scala/org/apache/kyuubi/plugin/spark/authz/ranger/RangerDelegationTokenProvider.scala

@@ -0,0 +1,86 @@
+/*
+ * 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.plugin.spark.authz.ranger
+
+import scala.util.control.NonFatal
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.security.{Credentials, UserGroupInformation}
+import org.apache.kyuubi.plugin.spark.authz.util.DTokenUtils
+import org.apache.ranger.admin.client.RangerAdminRESTClient
+import org.apache.spark.SparkConf
+import org.apache.spark.security.HadoopDelegationTokenProvider
+import org.slf4j.LoggerFactory
+
+class RangerDelegationTokenProvider extends HadoopDelegationTokenProvider {
+
+  private val LOG = LoggerFactory.getLogger(getClass)
+
+  private val PROPERTY_PREFIX = "ranger.plugin.spark"
+
+  private lazy val adminClient = {
+    val conf = SparkRangerAdminPlugin.getRangerConf
+    val serviceName = conf.get(PROPERTY_PREFIX + ".service.name", "spark")
+    val client = new RangerAdminRESTClient()
+    client.init(serviceName, "sparkSubmit", PROPERTY_PREFIX, conf)
+    client
+  }
+
+  override def serviceName: String = "ranger"
+
+  override def delegationTokensRequired(
+      sparkConf: SparkConf,
+      hadoopConf: Configuration): Boolean = {
+    UserGroupInformation.isSecurityEnabled
+  }
+
+  override def obtainDelegationTokens(
+      hadoopConf: Configuration,
+      sparkConf: SparkConf,
+      creds: Credentials): Option[Long] = {
+    try {
+      val rangerUrl = SparkRangerAdminPlugin.getRangerConf
+        .get(PROPERTY_PREFIX + ".policy.rest.url")
+
+      if (rangerUrl == null || rangerUrl.isEmpty) {
+        LOG.warn("Cannot obtain Ranger delegation token: " +
+          PROPERTY_PREFIX + ".policy.rest.url is not configured")
+        return None
+      }
+
+      val renewer = DTokenUtils.getTokenRenewer(sparkConf, hadoopConf)
+
+      LOG.info(s"Obtaining Ranger delegation token from $rangerUrl, renewer=$renewer")
+
+      val token = adminClient.getDelegationToken(renewer)
+
+      creds.addToken(token.getService, token)
+
+      val ident = token.decodeIdentifier()
+      LOG.info(s"Obtained Ranger delegation token, " +
+        s"owner=${ident.getUser}, renewer=$renewer, maxDate=${ident.getMaxDate}")
+
+      None
+    } catch {
+      case NonFatal(e) =>
+        LOG.warn(s"Failed to obtain Ranger delegation token. " +
+          s"If Ranger is not used, set spark.security.credentials.ranger.enabled=false", e)
+        None
+    }
+  }
+
+}

extensions/spark/kyuubi-spark-authz/src/main/scala/org/apache/kyuubi/plugin/spark/authz/ranger/SolrDelegationTokenProvider.scala

@@ -0,0 +1,130 @@
+/*
+ * 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.plugin.spark.authz.ranger
+
+import java.util.Properties
+import scala.jdk.CollectionConverters._
+import scala.util.control.NonFatal
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.security.{Credentials, UserGroupInformation}
+import org.apache.kyuubi.plugin.spark.authz.util.DTokenUtils
+import org.apache.ranger.audit.utils.InMemoryJAASConfiguration
+import org.apache.solr.client.solrj.impl.{CloudSolrClient, HttpClientUtil, HttpSolrClient, Krb5HttpClientBuilder}
+import org.apache.solr.client.solrj.request.DelegationTokenRequest
+import org.apache.solr.hadoop.SolrDelegationTokenUtil
+import org.apache.spark.SparkConf
+import org.apache.spark.security.HadoopDelegationTokenProvider
+import org.slf4j.LoggerFactory
+
+class SolrDelegationTokenProvider extends HadoopDelegationTokenProvider {
+
+  private val LOG = LoggerFactory.getLogger(getClass)
+
+  private val AUDIT_DEST_PREFIX = "xasecure.audit.destination.solr"
+  private val PROP_JAVA_SECURITY_AUTH_LOGIN_CONFIG = "java.security.auth.login.config"
+
+  override def serviceName: String = "solr"
+
+  override def delegationTokensRequired(
+      sparkConf: SparkConf,
+      hadoopConf: Configuration): Boolean = {
+    UserGroupInformation.isSecurityEnabled && {
+      try {
+        val conf = SparkRangerAdminPlugin.getRangerConf
+        val enabled = conf.get(AUDIT_DEST_PREFIX)
+        enabled != null &&
+          (enabled.equalsIgnoreCase("true") || enabled.equalsIgnoreCase("enabled"))
+      } catch {
+        case NonFatal(_) => false
+      }
+    }
+  }
+
+  override def obtainDelegationTokens(
+      hadoopConf: Configuration,
+      sparkConf: SparkConf,
+      creds: Credentials): Option[Long] = {
+    try {
+      val rangerConf = SparkRangerAdminPlugin.getRangerConf
+
+      val urls = rangerConf.get(AUDIT_DEST_PREFIX + ".urls")
+      val zkHosts = rangerConf.get(AUDIT_DEST_PREFIX + ".zookeepers")
+
+      val hasUrls = urls != null && urls.trim.nonEmpty && !urls.equalsIgnoreCase("NONE")
+      val hasZk = zkHosts != null && zkHosts.trim.nonEmpty && !zkHosts.equalsIgnoreCase("NONE")
+
+      if (!hasUrls && !hasZk) {
+        LOG.warn("Cannot obtain Solr delegation token: " +
+          s"neither $AUDIT_DEST_PREFIX.urls nor $AUDIT_DEST_PREFIX.zookeepers " +
+          "is configured in Ranger audit config")
+        return None
+      }
+
+      val renewer = DTokenUtils.getTokenRenewer(sparkConf, hadoopConf)
+      LOG.info(s"Obtaining Solr delegation token, urls=$urls, zkHosts=$zkHosts, renewer=$renewer")
+
+      // Set up JAAS from Ranger audit config (same as SolrAuditDestination.init())
+      val props = new Properties()
+      rangerConf.iterator().asScala.foreach(e => props.setProperty(e.getKey, e.getValue))
+
+      if (System.getProperty(PROP_JAVA_SECURITY_AUTH_LOGIN_CONFIG) == null) {
+        val forceInMemory = rangerConf.getBoolean(
+          AUDIT_DEST_PREFIX + ".force.use.inmemory.jaas.config", false)
+        if (forceInMemory) {
+          System.setProperty(PROP_JAVA_SECURITY_AUTH_LOGIN_CONFIG, "/dev/null")
+        }
+      }
+      InMemoryJAASConfiguration.init(props)
+
+      val krbBuild = new Krb5HttpClientBuilder()
+      HttpClientUtil.setHttpClientBuilder(krbBuild.getBuilder)
+
+      val client = if (hasZk) {
+        val zkHostsList = java.util.Arrays.asList(zkHosts.split(",").map(_.trim): _*)
+        new CloudSolrClient.Builder(zkHostsList, java.util.Optional.empty[String]()).build()
+      } else {
+        new HttpSolrClient.Builder(urls.split(",")(0).trim).build()
+      }
+
+      try {
+        val request = new DelegationTokenRequest.Get(renewer)
+        val response = request.process(client)
+        val dtString = response.getDelegationToken
+
+        if (dtString == null) {
+          LOG.warn("Solr returned null delegation token")
+          return None
+        }
+
+        val serviceUrl = if (hasUrls) urls.split(",")(0).trim else zkHosts
+        val token = SolrDelegationTokenUtil.createHadoopToken(dtString, serviceUrl)
+        creds.addToken(token.getService, token)
+
+        LOG.info("Obtained Solr delegation token successfully")
+        None
+      } finally {
+        client.close()
+      }
+    } catch {
+      case NonFatal(e) =>
+        LOG.warn("Failed to obtain Solr delegation token. " +
+          "If Solr audit is not used, set spark.security.credentials.solr.enabled=false", e)
+        None
+    }
+  }
+}

extensions/spark/kyuubi-spark-authz/src/main/scala/org/apache/kyuubi/plugin/spark/authz/util/DTokenUtils.scala

@@ -0,0 +1,37 @@
+/*
+ * 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.plugin.spark.authz.util
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.security.{SecurityUtil, UserGroupInformation}
+import org.apache.spark.SparkConf
+
+object DTokenUtils {
+  def getTokenRenewer(
+                       sparkConf: SparkConf,
+                       hadoopConf: Configuration): String = {
+    val master = sparkConf.get("spark.master", "")
+    if (master.contains("yarn")) {
+      val rmPrincipal = hadoopConf.get("yarn.resourcemanager.principal")
+      if (rmPrincipal != null && rmPrincipal.nonEmpty) {
+        return SecurityUtil.getServerPrincipal(rmPrincipal, null: String)
+      }
+    }
+    UserGroupInformation.getCurrentUser.getUserName
+  }
+}