[KYUUBI apache#4504] [Authz] [Bug] Fix source table privilege require…

…ment when querying permanent view in Spark 3.1 and below

### _Why are the changes needed?_

To fix apache#4504.
- add ut for querying the second column of permenant view with multi-column source table (in purpose for reproducing cases for Spark 3.1 and below)
- make user RuleAuthorization check only once, by marking `KYUUBI_AUTHZ_TAG` tag on plan's children and itself , to prevent penetration checks to source table of perm view

### _How was this patch tested?_
- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [x] [Run test](https://kyuubi.readthedocs.io/en/master/develop_tools/testing.html#running-tests) locally before make a pull request

Closes apache#4529 from bowenliang123/4504-authz-permview.

Closes apache#4504

b5f9bed [liangbowen] minor update
3efba2d [liangbowen] update ut and policy
a7584d9 [liangbowen] update
c0c1155 [liangbowen] update
ee208d7 [liangbowen] update
3c7804e [liangbowen] fix ut
5ab0d1a [liangbowen] check RuleAuthorization run only once

Authored-by: liangbowen <liangbowen@gf.com.cn>
Signed-off-by: liangbowen <liangbowen@gf.com.cn>

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

@@ -27,24 +27,23 @@ import org.apache.spark.sql.catalyst.trees.TreeNodeTag
 
 import org.apache.kyuubi.plugin.spark.authz._
 import org.apache.kyuubi.plugin.spark.authz.ObjectType._
-import org.apache.kyuubi.plugin.spark.authz.ranger.RuleAuthorization.KYUUBI_AUTHZ_TAG
+import org.apache.kyuubi.plugin.spark.authz.ranger.RuleAuthorization._
 import org.apache.kyuubi.plugin.spark.authz.ranger.SparkRangerAdminPlugin._
-import org.apache.kyuubi.plugin.spark.authz.util.AuthZUtils._;
+import org.apache.kyuubi.plugin.spark.authz.util.AuthZUtils._
 class RuleAuthorization(spark: SparkSession) extends Rule[LogicalPlan] {
-  override def apply(plan: LogicalPlan): LogicalPlan = plan match {
-    case p if !plan.getTagValue(KYUUBI_AUTHZ_TAG).contains(true) =>
-      RuleAuthorization.checkPrivileges(spark, p)
-      p.setTagValue(KYUUBI_AUTHZ_TAG, true)
-      p
-    case p => p // do nothing if checked privileges already.
+  override def apply(plan: LogicalPlan): LogicalPlan = {
+    plan match {
+      case plan if isAuthChecked(plan) => plan // do nothing if checked privileges already.
+      case p => checkPrivileges(spark, p)
+    }
   }
 }
 
 object RuleAuthorization {
 
   val KYUUBI_AUTHZ_TAG = TreeNodeTag[Boolean]("__KYUUBI_AUTHZ_TAG")
 
-  def checkPrivileges(spark: SparkSession, plan: LogicalPlan): Unit = {
+  private def checkPrivileges(spark: SparkSession, plan: LogicalPlan): LogicalPlan = {
     val auditHandler = new SparkRangerAuditHandler
     val ugi = getAuthzUgi(spark.sparkContext)
     val (inputs, outputs, opType) = PrivilegesBuilder.build(plan, spark)
@@ -94,5 +93,17 @@ object RuleAuthorization {
         verify(Seq(req), auditHandler)
       }
     }
+    markAuthChecked(plan)
+  }
+
+  private def markAuthChecked(plan: LogicalPlan): LogicalPlan = {
+    plan.transformUp { case p =>
+      p.setTagValue(KYUUBI_AUTHZ_TAG, true)
+      p
+    }
+  }
+
+  private def isAuthChecked(plan: LogicalPlan): Boolean = {
+    plan.find(_.getTagValue(KYUUBI_AUTHZ_TAG).contains(true)).nonEmpty
   }
 }

extensions/spark/kyuubi-spark-authz/src/test/resources/sparkSql_hive_jenkins.json

@@ -1151,6 +1151,67 @@
       "guid": "b3f1f1e0-2bd6-4b20-8a32-a531006ae151",
       "isEnabled": true,
       "version": 1
+    },
+    {
+      "service": "hive_jenkins",
+      "name": "someone_access_perm_view",
+      "policyType": 0,
+      "policyPriority": 0,
+      "description": "",
+      "isAuditEnabled": true,
+      "resources": {
+        "database": {
+          "values": [
+            "default"
+          ],
+          "isExcludes": false,
+          "isRecursive": false
+        },
+        "column": {
+          "values": [
+            "*"
+          ],
+          "isExcludes": false,
+          "isRecursive": false
+        },
+        "table": {
+          "values": [
+            "perm_view"
+          ],
+          "isExcludes": false,
+          "isRecursive": false
+        }
+      },
+      "policyItems": [
+        {
+          "accesses": [
+            {
+              "type": "select",
+              "isAllowed": true
+            }
+          ],
+          "users": [
+            "user_perm_view_only"
+          ],
+          "groups": [],
+          "conditions": [],
+          "delegateAdmin": false
+        }
+      ],
+      "denyPolicyItems": [],
+      "allowExceptions": [],
+      "denyExceptions": [],
+      "dataMaskPolicyItems": [],
+      "rowFilterPolicyItems": [],
+      "options": {},
+      "validitySchedules": [],
+      "policyLabels": [
+        ""
+      ],
+      "id": 123,
+      "guid": "2fb6099d-e421-41df-9d24-f2f47bed618e",
+      "isEnabled": true,
+      "version": 5
     }
   ],
   "serviceDef": {

extensions/spark/kyuubi-spark-authz/src/test/scala/org/apache/kyuubi/plugin/spark/authz/ranger/RangerSparkExtensionSuite.scala

@@ -521,27 +521,61 @@ class HiveCatalogRangerSparkExtensionSuite extends RangerSparkExtensionSuite {
   }
 
   test("[KYUUBI #3326] check persisted view and skip shadowed table") {
+    val db1 = "default"
     val table = "hive_src"
     val permView = "perm_view"
-    val db1 = "default"
-    val db2 = "db2"
 
     withCleanTmpResources(Seq(
       (s"$db1.$table", "table"),
-      (s"$db2.$permView", "view"),
-      (db2, "database"))) {
-      doAs("admin", sql(s"CREATE TABLE IF NOT EXISTS $db1.$table (id int)"))
-
-      doAs("admin", sql(s"CREATE DATABASE IF NOT EXISTS $db2"))
-      doAs("admin", sql(s"CREATE VIEW $db2.$permView AS SELECT * FROM $table"))
+      (s"$db1.$permView", "view"))) {
+      doAs("admin", sql(s"CREATE TABLE IF NOT EXISTS $db1.$table (id int, name string)"))
+      doAs("admin", sql(s"CREATE VIEW $db1.$permView AS SELECT * FROM $db1.$table"))
 
+      // KYUUBI #3326: with no privileges to the permanent view or the source table
       val e1 = intercept[AccessControlException](
-        doAs("someone", sql(s"select * from $db2.$permView")).show(0))
+        doAs(
+          "someone", {
+            sql(s"select * from $db1.$permView").collect()
+          }))
+      if (isSparkV31OrGreater) {
+        assert(e1.getMessage.contains(s"does not have [select] privilege on [$db1/$permView/id]"))
+      } else {
+        assert(e1.getMessage.contains(s"does not have [select] privilege on [$db1/$table/id]"))
+      }
+    }
+  }
+
+  test("KYUUBI #4504: query permanent view with privilege to permanent view only") {
+    val db1 = "default"
+    val table = "hive_src"
+    val permView = "perm_view"
+    val userPermViewOnly = "user_perm_view_only"
+
+    withCleanTmpResources(Seq(
+      (s"$db1.$table", "table"),
+      (s"$db1.$permView", "view"))) {
+      doAs("admin", sql(s"CREATE TABLE IF NOT EXISTS $db1.$table (id int, name string)"))
+      doAs("admin", sql(s"CREATE VIEW $db1.$permView AS SELECT * FROM $db1.$table"))
+
+      // query all columns of the permanent view
+      // with access privileges to the permanent view but no privilege to the source table
+      val sql1 = s"SELECT * FROM $db1.$permView"
       if (isSparkV31OrGreater) {
-        assert(e1.getMessage.contains(s"does not have [select] privilege on [$db2/$permView/id]"))
+        doAs(userPermViewOnly, { sql(sql1).collect() })
       } else {
+        val e1 = intercept[AccessControlException](doAs(userPermViewOnly, { sql(sql1).collect() }))
         assert(e1.getMessage.contains(s"does not have [select] privilege on [$db1/$table/id]"))
       }
+
+      // query the second column of permanent view with multiple columns
+      // with access privileges to the permanent view but no privilege to the source table
+      val sql2 = s"SELECT name FROM $db1.$permView"
+      if (isSparkV31OrGreater) {
+        doAs(userPermViewOnly, { sql(sql2).collect() })
+      } else {
+        val e2 = intercept[AccessControlException](doAs(userPermViewOnly, { sql(sql2).collect() }))
+        assert(e2.getMessage.contains(s"does not have [select] privilege on [$db1/$table/name]"))
+      }
     }
   }