zhangxinyao88 commented on code in PR #7437:
URL: https://github.com/apache/kyuubi/pull/7437#discussion_r4129586915
##########
extensions/spark/kyuubi-spark-authz/src/main/scala/org/apache/kyuubi/plugin/spark/authz/ranger/SparkRangerAdminPlugin.scala:
##########
@@ -17,114 +17,354 @@
package org.apache.kyuubi.plugin.spark.authz.ranger
+import java.util.Properties
+
import scala.collection.JavaConverters._
import scala.collection.mutable.{ArrayBuffer, LinkedHashMap}
+import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.util.ShutdownHookManager
-import org.apache.ranger.plugin.policyengine.RangerAccessRequest
-import org.apache.ranger.plugin.service.RangerBasePlugin
+import org.apache.ranger.authz.api.{RangerAuthorizer, RangerAuthorizerFactory}
+import org.apache.ranger.authz.model._
+import
org.apache.ranger.authz.model.RangerAccessContext.CONTEXT_INFO_CLUSTER_NAME
+import org.apache.ranger.authz.model.RangerAuthzResult.AccessDecision
import org.slf4j.LoggerFactory
import org.apache.kyuubi.plugin.spark.authz.AccessControlException
+import org.apache.kyuubi.plugin.spark.authz.ObjectType
+import org.apache.kyuubi.plugin.spark.authz.ranger.AccessType._
-object SparkRangerAdminPlugin extends RangerBasePlugin("spark", "sparkSql")
- with RangerConfigProvider {
+/**
+ * The entry point of Ranger authorization for Spark SQL, which delegates
+ * authorization requests to a Ranger authorizer:
+ * <ul>
+ * <li>org.apache.ranger.authz.remote.RangerRemoteAuthorizer (default):
+ * sends requests to a Ranger PDP server via REST APIs, with a thin
client
+ * that does not download policies to the client side.</li>
+ * <li>org.apache.ranger.authz.embedded.RangerEmbeddedAuthorizer:
+ * evaluates requests locally against the policies pulled from
+ * the Ranger admin server, as the previous Ranger plugin did.</li>
+ * </ul>
+ *
+ * The authorizer implementation is selected by `ranger.authorizer.impl.class`.
+ */
+object SparkRangerAdminPlugin {
final private val LOG = LoggerFactory.getLogger(getClass)
+ /**
+ * The service type of the Spark SQL service definition registered in
+ * Ranger admin.
+ */
+ final val SERVICE_TYPE: String = "spark"
+
+ final private val APP_TYPE: String = "sparkSql"
+
+ final private val KEY_IMPL_CLASS = "ranger.authorizer.impl.class"
+ final private val KEY_PDP_URL = "ranger.authz.remote.pdp.url"
+ final private val KEY_SERVICE_NAME =
s"ranger.plugin.$SERVICE_TYPE.service.name"
+ final private val KEY_CLUSTER_NAME =
s"ranger.plugin.$SERVICE_TYPE.access.cluster.name"
+ final private val KEY_AUTHORIZE_IN_SINGLE_CALL =
+ s"ranger.plugin.$SERVICE_TYPE.authorize.in.single.call"
+
+ final private val REMOTE_AUTHORIZER_CLASS: String =
+ classOf[org.apache.ranger.authz.remote.RangerRemoteAuthorizer].getName
+
+ /**
+ * The security-critical configurations which are read from the Ranger
configuration
+ * resources only: JVM system properties can neither set nor override them,
so a tenant
+ * who can control the JVM options of the engine process cannot point the
authorization
+ * to a server they control or bypass the policies.
+ */
+ final private val ADMIN_ONLY_KEYS: Seq[String] =
+ Seq(KEY_IMPL_CLASS, KEY_PDP_URL, KEY_SERVICE_NAME)
+
+ final private val ADMIN_ONLY_KEY_PREFIXES: Seq[String] = Seq(
+ "ranger.authz.remote.authn.",
+ "ranger.authz.remote.header.",
+ "ranger.authz.remote.ssl.")
+
+ final private val CONFIG_PREFIXES = Seq("ranger.", "xasecure.")
+
+ final private val CONFIG_RESOURCES = Seq(
+ s"ranger-$SERVICE_TYPE-security.xml",
+ s"ranger-$SERVICE_TYPE-audit.xml")
+
+ private var authorizer: RangerAuthorizer = null
+
+ /**
+ * The Ranger configurations loaded from the Hadoop configuration resources,
+ * e.g. ranger-spark-security.xml.
+ */
+ private[ranger] var config: Configuration = null
+
/**
* For a Spark SQL query, it may contain 0 or more privilege objects to
verify, e.g. a typical
* JOIN operator may have two tables and their columns to verify.
*
* This configuration controls whether to verify the privilege objects in
single call or
* to verify them one by one.
*/
- def authorizeInSingleCall: Boolean = getRangerConf.getBoolean(
- s"ranger.plugin.${getServiceType}.authorize.in.single.call",
- false)
+ def authorizeInSingleCall: Boolean =
config.getBoolean(KEY_AUTHORIZE_IN_SINGLE_CALL, false)
- /**
- * This configuration controls whether to override user's usergroups
- * by the mapping fetched from Ranger's UserStore.
- *
- * It relies on Ranger's UserStore is a feature supported since Ranger 2.1.
- *
- * If true, user bound usergroups will be looked up in in Ranger's UserStore
- * and the usergroups of AccessRequest is overriden.
- *
- * Please make sure configs in Ranger set properly:
- * 1. set `ranger.plugin.spark.enable.implicit.userstore.enricher` to true
- * 2. set cache path for UserStore in `ranger.plugin.hive.policy.cache.dir`
- * 3. at least one condition of policies containing scripts, e.g.
{{USER.attr}} in row-filter
- */
- def useUserGroupsFromUserStoreEnabled: Boolean = getRangerConf.getBoolean(
- s"ranger.plugin.$getServiceType.use.usergroups.from.userstore.enabled",
- false)
+ def getServiceType: String = SERVICE_TYPE
+
+ private def serviceName: String = config.get(KEY_SERVICE_NAME)
/**
- * plugin initialization
+ * authorizer initialization
* with cleanup shutdown hook registered
*/
def initialize(): Unit = {
- this.init()
- registerCleanupShutdownHook(this)
+ ensureConfig()
+ val props = new Properties
+ config.iterator.asScala
+ .filter(entry => CONFIG_PREFIXES.exists(entry.getKey.startsWith))
+ .foreach(entry => props.put(entry.getKey, entry.getValue))
+ System.getProperties.asScala
+ .filter { case (key, _) =>
+ CONFIG_PREFIXES.exists(key.startsWith) && !isAdminOnlyKey(key)
+ }
+ .foreach { case (key, value) => props.put(key, value) }
+ validateAdminOnlyKeys(props)
+ initialize(props)
+ }
+
+ private def isAdminOnlyKey(key: String): Boolean =
+ ADMIN_ONLY_KEYS.contains(key) ||
ADMIN_ONLY_KEY_PREFIXES.exists(key.startsWith)
+
+ /**
+ * Fails fast when the Ranger configuration resources lack a configuration
the
+ * authorizer needs: the authorizer implementation defaults to the Ranger
PDP mode,
+ * which requires the Ranger PDP server address. The configuration cannot
fall back
+ * to JVM system properties, which tenant users may control.
+ */
+ private def validateAdminOnlyKeys(props: Properties): Unit = {
+ val implClass = props.getProperty(KEY_IMPL_CLASS, REMOTE_AUTHORIZER_CLASS)
Review Comment:
Ranger's factory defaults to embedded when `ranger.authorizer.impl.class` is
unset. Are we setting it to remote elsewhere? Otherwise the documented PDP
setup will try to load embedded, which is test-scoped and absent from the
shaded jar.
##########
extensions/spark/kyuubi-spark-authz/src/main/scala/org/apache/kyuubi/plugin/spark/authz/ranger/SparkRangerAdminPlugin.scala:
##########
@@ -17,114 +17,354 @@
package org.apache.kyuubi.plugin.spark.authz.ranger
+import java.util.Properties
+
import scala.collection.JavaConverters._
import scala.collection.mutable.{ArrayBuffer, LinkedHashMap}
+import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.util.ShutdownHookManager
-import org.apache.ranger.plugin.policyengine.RangerAccessRequest
-import org.apache.ranger.plugin.service.RangerBasePlugin
+import org.apache.ranger.authz.api.{RangerAuthorizer, RangerAuthorizerFactory}
+import org.apache.ranger.authz.model._
+import
org.apache.ranger.authz.model.RangerAccessContext.CONTEXT_INFO_CLUSTER_NAME
+import org.apache.ranger.authz.model.RangerAuthzResult.AccessDecision
import org.slf4j.LoggerFactory
import org.apache.kyuubi.plugin.spark.authz.AccessControlException
+import org.apache.kyuubi.plugin.spark.authz.ObjectType
+import org.apache.kyuubi.plugin.spark.authz.ranger.AccessType._
-object SparkRangerAdminPlugin extends RangerBasePlugin("spark", "sparkSql")
- with RangerConfigProvider {
+/**
+ * The entry point of Ranger authorization for Spark SQL, which delegates
+ * authorization requests to a Ranger authorizer:
+ * <ul>
+ * <li>org.apache.ranger.authz.remote.RangerRemoteAuthorizer (default):
+ * sends requests to a Ranger PDP server via REST APIs, with a thin
client
+ * that does not download policies to the client side.</li>
+ * <li>org.apache.ranger.authz.embedded.RangerEmbeddedAuthorizer:
+ * evaluates requests locally against the policies pulled from
+ * the Ranger admin server, as the previous Ranger plugin did.</li>
+ * </ul>
+ *
+ * The authorizer implementation is selected by `ranger.authorizer.impl.class`.
+ */
+object SparkRangerAdminPlugin {
final private val LOG = LoggerFactory.getLogger(getClass)
+ /**
+ * The service type of the Spark SQL service definition registered in
+ * Ranger admin.
+ */
+ final val SERVICE_TYPE: String = "spark"
+
+ final private val APP_TYPE: String = "sparkSql"
+
+ final private val KEY_IMPL_CLASS = "ranger.authorizer.impl.class"
+ final private val KEY_PDP_URL = "ranger.authz.remote.pdp.url"
+ final private val KEY_SERVICE_NAME =
s"ranger.plugin.$SERVICE_TYPE.service.name"
+ final private val KEY_CLUSTER_NAME =
s"ranger.plugin.$SERVICE_TYPE.access.cluster.name"
+ final private val KEY_AUTHORIZE_IN_SINGLE_CALL =
+ s"ranger.plugin.$SERVICE_TYPE.authorize.in.single.call"
+
+ final private val REMOTE_AUTHORIZER_CLASS: String =
+ classOf[org.apache.ranger.authz.remote.RangerRemoteAuthorizer].getName
+
+ /**
+ * The security-critical configurations which are read from the Ranger
configuration
+ * resources only: JVM system properties can neither set nor override them,
so a tenant
+ * who can control the JVM options of the engine process cannot point the
authorization
+ * to a server they control or bypass the policies.
+ */
+ final private val ADMIN_ONLY_KEYS: Seq[String] =
+ Seq(KEY_IMPL_CLASS, KEY_PDP_URL, KEY_SERVICE_NAME)
+
+ final private val ADMIN_ONLY_KEY_PREFIXES: Seq[String] = Seq(
+ "ranger.authz.remote.authn.",
+ "ranger.authz.remote.header.",
+ "ranger.authz.remote.ssl.")
+
+ final private val CONFIG_PREFIXES = Seq("ranger.", "xasecure.")
+
+ final private val CONFIG_RESOURCES = Seq(
+ s"ranger-$SERVICE_TYPE-security.xml",
+ s"ranger-$SERVICE_TYPE-audit.xml")
+
+ private var authorizer: RangerAuthorizer = null
+
+ /**
+ * The Ranger configurations loaded from the Hadoop configuration resources,
+ * e.g. ranger-spark-security.xml.
+ */
+ private[ranger] var config: Configuration = null
+
/**
* For a Spark SQL query, it may contain 0 or more privilege objects to
verify, e.g. a typical
* JOIN operator may have two tables and their columns to verify.
*
* This configuration controls whether to verify the privilege objects in
single call or
* to verify them one by one.
*/
- def authorizeInSingleCall: Boolean = getRangerConf.getBoolean(
- s"ranger.plugin.${getServiceType}.authorize.in.single.call",
- false)
+ def authorizeInSingleCall: Boolean =
config.getBoolean(KEY_AUTHORIZE_IN_SINGLE_CALL, false)
- /**
- * This configuration controls whether to override user's usergroups
- * by the mapping fetched from Ranger's UserStore.
- *
- * It relies on Ranger's UserStore is a feature supported since Ranger 2.1.
- *
- * If true, user bound usergroups will be looked up in in Ranger's UserStore
- * and the usergroups of AccessRequest is overriden.
- *
- * Please make sure configs in Ranger set properly:
- * 1. set `ranger.plugin.spark.enable.implicit.userstore.enricher` to true
- * 2. set cache path for UserStore in `ranger.plugin.hive.policy.cache.dir`
- * 3. at least one condition of policies containing scripts, e.g.
{{USER.attr}} in row-filter
- */
- def useUserGroupsFromUserStoreEnabled: Boolean = getRangerConf.getBoolean(
- s"ranger.plugin.$getServiceType.use.usergroups.from.userstore.enabled",
- false)
+ def getServiceType: String = SERVICE_TYPE
+
+ private def serviceName: String = config.get(KEY_SERVICE_NAME)
/**
- * plugin initialization
+ * authorizer initialization
* with cleanup shutdown hook registered
*/
def initialize(): Unit = {
- this.init()
- registerCleanupShutdownHook(this)
+ ensureConfig()
+ val props = new Properties
+ config.iterator.asScala
+ .filter(entry => CONFIG_PREFIXES.exists(entry.getKey.startsWith))
+ .foreach(entry => props.put(entry.getKey, entry.getValue))
+ System.getProperties.asScala
+ .filter { case (key, _) =>
+ CONFIG_PREFIXES.exists(key.startsWith) && !isAdminOnlyKey(key)
Review Comment:
Can we protect the embedded policy source settings too? A tenant could
override `ranger.plugin.spark.policy.source.impl` through JVM properties and
load policies from a source they control.
##########
extensions/spark/kyuubi-spark-authz/src/main/scala/org/apache/kyuubi/plugin/spark/authz/ranger/AccessResource.scala:
##########
@@ -18,27 +18,167 @@
package org.apache.kyuubi.plugin.spark.authz.ranger
import java.io.File
-import java.util
-import scala.language.implicitConversions
+import scala.collection.JavaConverters._
-import org.apache.ranger.plugin.policyengine.RangerAccessResourceImpl
+import org.apache.commons.lang3.StringUtils
+import org.apache.ranger.authz.model.RangerResourceInfo
-import org.apache.kyuubi.plugin.spark.authz.{ObjectType, PrivilegeObject}
+import org.apache.kyuubi.plugin.spark.authz.{AccessControlException,
ObjectType, PrivilegeObject}
import org.apache.kyuubi.plugin.spark.authz.ObjectType._
import org.apache.kyuubi.plugin.spark.authz.OperationType.OperationType
-class AccessResource private (val objectType: ObjectType, val catalog:
Option[String])
- extends RangerAccessResourceImpl {
- implicit def asString(obj: Object): String = if (obj != null)
obj.asInstanceOf[String] else null
- def getDatabase: String = getValue("database")
- def getUdf: String = getValue("udf")
- def getTable: String = getValue("table")
- def getColumn: String = getValue("column")
+/**
+ * A privilege object to authorize, which is converted to a Ranger resource
+ * (e.g. "table:default/src" or "column:default/src/id") in requests to
+ * the Ranger authorizer.
+ *
+ * @param objectType the type of the object
+ * @param database the database name, or null if not applicable
+ * @param table the table name, or null if not applicable
+ * @param column the column names joined by comma, or null if not
applicable
+ * @param udf the function name, or null if not applicable
+ * @param uri the uri path, or null if not applicable
+ * @param owner the owner of the object, if any
+ * @param catalog the catalog name, if any
+ */
+case class AccessResource private[ranger] (
+ objectType: ObjectType,
+ database: String,
+ table: String,
+ column: String,
+ udf: String,
+ uri: String,
+ owner: Option[String],
+ catalog: Option[String]) {
+
+ def getDatabase: String = database
+ def getUdf: String = udf
+ def getTable: String = table
+ def getColumn: String = column
+
def getColumns: Seq[String] = {
- val columnStr = getColumn
- if (columnStr == null) Nil else columnStr.split(",").filter(_.nonEmpty)
+ if (column == null) Nil else column.split(",").filter(_.nonEmpty)
}
+
+ def getOwnerUser: String = owner.orNull
+
+ /**
+ * The path-like representation of this resource used in error messages,
+ * e.g. "default/src" for a table, "default/src/id" for a column.
+ */
+ def getAsString: String = objectType match {
+ case COLUMN =>
+ Seq(database, table, column).filter(_ != null).mkString("/")
+ case FUNCTION =>
+ Seq(database, udf).filter(_ != null).mkString("/")
+ case URI =>
+ // the uri is matched against both the exact path and the path with a
+ // trailing slash, as the legacy plugin did
+ val path = Option(uri).map(_.stripSuffix(File.separator)).getOrElse("")
+ s"[$path, $path/]"
+ case _ =>
+ Seq(database, table).filter(_ != null).mkString("/")
+ }
+
+ private[ranger] def toResourceInfos: Seq[RangerResourceInfo] = {
+ val attributes = owner.map(o =>
java.util.Collections.singletonMap("OWNER", o: AnyRef)).orNull
+ objectType match {
+ case DATABASE =>
+ Seq(new RangerResourceInfo(
+ s"database:${requireRrnComponent(database, "database")}",
+ null,
+ null,
+ attributes))
+ case FUNCTION =>
+ // An unqualified function reference (e.g. a built-in or temporary
function) has no
+ // database. The legacy plugin left the database blank, and blank
values in the legacy
+ // resource matched the wildcard values in policies, so keep the
wildcard marker for
+ // the blank database. Blank names in other resource levels indicate a
broken command
+ // extraction, which matched no policy in the legacy resource, so deny
them.
+ val db = if (StringUtils.isBlank(database)) "*" else
escapeRrnMetaChars(database)
+ Seq(new RangerResourceInfo(
+ s"udf:$db/${requireRrnComponent(udf, "udf")}",
+ null,
+ null,
+ attributes))
+ case COLUMN =>
+ val columns = getColumns
+ // all the column requests need the database and the table
+ val parent =
+ s"column:${requireRrnComponent(database, "database")}" +
+ s"/${requireRrnComponent(table, "table")}"
+ if (columns.length == 1) {
+ Seq(new RangerResourceInfo(
+ s"$parent/${requireRrnComponent(columns.head, "column")}",
+ null,
+ null,
+ attributes))
+ } else if (columns.isEmpty) {
+ Seq(new RangerResourceInfo(parent, null, null, attributes))
+ } else {
+ val subResources = columns
+ .map(col => s"column:${requireRrnComponent(col, "column")}")
+ .toSet.asJava
+ Seq(new RangerResourceInfo(parent, subResources, null, attributes))
+ }
+ case URI =>
+ // Url policies may be written with or without a trailing slash, and
the legacy
+ // plugin matched a uri against both the exact path and the path with
a trailing
+ // slash. The RRN request carries a single resource value set, so the
two variants
+ // are returned and authorized as alternatives by separate requests.
+ val path = requireRrnComponent(
+ Option(uri).map(_.stripSuffix(File.separator)).orNull,
+ "uri")
+ Seq(
+ new RangerResourceInfo(s"url:$path", null, null, attributes),
+ new RangerResourceInfo(s"url:$path/", null, null, attributes))
+ case VIEW =>
+ // A local temporary view has no database in the SHOW TABLES output.
The legacy
+ // plugin left the database blank, and blank values in the legacy
resource matched
+ // the wildcard values in policies, so keep the wildcard marker for
the blank
+ // database, as the unqualified function reference case does.
+ val db = if (StringUtils.isBlank(database)) "*" else
escapeRrnMetaChars(database)
+ Seq(new RangerResourceInfo(
+ s"table:$db/${requireRrnComponent(table, "table")}",
+ null,
+ null,
+ attributes))
+ case _ =>
+ Seq(new RangerResourceInfo(
+ s"table:${requireRrnComponent(database, "database")}" +
+ s"/${requireRrnComponent(table, "table")}",
+ null,
+ null,
+ attributes))
+ }
+ }
+
+ /**
+ * Escapes the RRN metacharacters in a resource name, so that a name
containing
+ * them is parsed as a single resource level instead of breaking the resource
+ * hierarchy, e.g. a table named "a/b" becomes "a\/b" in the resource name
+ * "table:default/a\/b". The escape format follows RangerResourceNameParser:
+ * a backslash escapes the next character, so "\\" stands for a literal
+ * backslash and "\/" stands for a literal separator.
+ */
+ private def escapeRrnMetaChars(value: String): String =
+ value.replace("\\", "\\\\").replace("/", "\\/")
+
+ /**
+ * Returns the RRN component for the given resource name, throwing an access
+ * control exception for a blank name. The RRN parser rejects blank resource
+ * values, and the legacy plugin had no value for a blank name, which matched
+ * no policy, so the access is denied rather than being evaluated against the
+ * wildcard marker.
+ */
+ private def requireRrnComponent(value: String, name: String): String =
+ if (StringUtils.isBlank(value)) {
Review Comment:
The Java 21 / Spark 4.0 CI job fails here in Paimon's `Producers` test: a
catalog-only operation has a blank database. Could we handle that case and add
a test?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
