[SC-5483] Fix CreateTempTable/ShowPermissions/ShowDatabases in CheckPermissions

## What changes were proposed in this pull request?
The CheckPermissions rule currently does not handle  the `ShowDatabasesCommand` and the `ShowPermissionsCommand` nodes, as a result the rule will throw a SecurityException when one of these commands is used. This PR adds these commands to the whitelist.

We also change the required permission for `CreateTempTableUsing` and the 'run-SQL-on-file' functionality, both now require a Select on an Anonymous File (a blanket permission). The reason for this is that both code paths allow a user to interact directly with the filesystem, and gives the user a way to circumvent ACLs defined on a table (by querying the files backing the table directly).

## How was this patch tested?
Added tests to `CheckPermissionRuleSuite`.

Author: Herman van Hovell <[email protected]>

Closes apache#156 from hvanhovell/SC-5483.

Author: hvanhovell

sql/core/src/main/antlr4/com/databricks/sql/acl/AclCommandBase.g4

@@ -39,6 +39,7 @@ securable
     | objectType=VIEW qualifiedName
     | objectType=FUNCTION qualifiedName
     | ANONYMOUS objectType=FUNCTION
+    | ANY objectType=FILE
     | objectType=TABLE? qualifiedName
     ;
 
@@ -58,7 +59,7 @@ quotedIdentifier
 
 nonReserved
     : ALTER | OWNER | TO | MSCK | REPAIR | PRIVILEGES | SHOW | GRANT | ON | ALL | WITH | OPTION |
-      REVOKE | FOR | FROM | CATALOG | DATABASE | TABLE | VIEW | FUNCTION | ANONYMOUS
+      REVOKE | FOR | FROM | CATALOG | DATABASE | TABLE | VIEW | FUNCTION | ANONYMOUS | FILE | ANY
     ;
 
 ALTER: 'ALTER';
@@ -82,6 +83,8 @@ TABLE: 'TABLE';
 VIEW: 'VIEW';
 FUNCTION: 'FUNCTION';
 ANONYMOUS: 'ANONYMOUS';
+FILE: 'FILE';
+ANY: 'ANY';
 
 STRING
     : '\'' ( ~('\''|'\\') | ('\\' .) )* '\''

sql/core/src/main/scala/com/databricks/sql/acl/AstCommandBuilder.scala

@@ -139,6 +139,8 @@ class AstCommandBuilder(client: AclClient)
         AnonymousFunction
       case AclCommandBaseParser.FUNCTION =>
         Function(visitFunctionIdentifier(ctx.qualifiedName))
+      case AclCommandBaseParser.FILE =>
+        AnyFile
       case _ =>
         throw new ParseException("Unknown Securable Object", ctx)
     }

sql/core/src/main/scala/com/databricks/sql/acl/CheckPermissions.scala

@@ -80,6 +80,8 @@ class CheckPermissions(catalog: PublicCatalog, aclClient: AclClient)
       case rel: CatalogRelation if isTempDb(rel.catalogTable.identifier.database) => Nil
       case rel: CatalogRelation => Seq(Request(Table(rel.catalogTable.identifier), Select))
 
+      case LogicalRelation(_: HadoopFsRelation, _, None) =>
+        Seq(Select(AnyFile))
       case LogicalRelation(_, _, None) => Nil
       case LogicalRelation(_, _, Some(tableId)) if isTempDb(tableId.identifier.database) => Nil
       case LogicalRelation(_, _, Some(tableId)) =>
@@ -131,7 +133,8 @@ class CheckPermissions(catalog: PublicCatalog, aclClient: AclClient)
     case create: CreateTable =>
       toSecurables(create.tableDesc.identifier.database).map(Create) ++
         create.query.toSeq.flatMap(queryToRequests)
-    case create: CreateTempViewUsing => Nil
+    case _: CreateTempViewUsing =>
+      Seq(Select(AnyFile))
     case create: CreateDataSourceTableCommand =>
       toSecurables(create.table.identifier.database).map(Create)
     case create: CreateTableCommand =>
@@ -192,7 +195,7 @@ class CheckPermissions(catalog: PublicCatalog, aclClient: AclClient)
     case drop: DropFunctionCommand =>
       toSecurables(FunctionIdentifier(drop.functionName, drop.databaseName)).map(Own)
 
-    case show: ShowDatabasesCommand =>
+    case _: ShowDatabasesCommand =>
       Seq(Request(Catalog, ReadMetadata))
     case show: ShowTablesCommand =>
       toSecurables(show.databaseName).map(ReadMetadata)

sql/core/src/main/scala/com/databricks/sql/acl/commands.scala

@@ -38,6 +38,8 @@ trait AclCommand extends RunnableCommand {
         Some(Function(FunctionIdentifier(fn, Option(qualified.database))))
       case Function(FunctionIdentifier(fn, Some(db))) if catalog.functionExists(db, fn) =>
         Some(securable)
+      case AnyFile =>
+        Some(securable)
       case _ => None
     }
   }

sql/core/src/main/scala/com/databricks/sql/acl/interfaces.scala

@@ -139,6 +139,8 @@ object Securable {
     parseString(rep) match {
       case Seq("ANONYMOUS_FUNCTION") =>
         AnonymousFunction
+      case Seq("ANY_FILE") =>
+        AnyFile
       case Seq("CATALOG", "default") =>
         Catalog
       case Seq("CATALOG", "default", "DATABASE", db) =>
@@ -203,6 +205,12 @@ case object AnonymousFunction extends SecurableFunction {
   override val name: String = "/" + typeName
 }
 
+case object AnyFile extends Securable {
+  override val key: Option[Nothing] = None
+  override def typeName: String = "ANY_FILE"
+  override val name: String = "/" + typeName
+}
+
 /**
  * The action that a [[Principal]] wants to execute on a [[Securable]]. An [[Action]] should always
  * be a verb.
@@ -275,12 +283,12 @@ object Action {
     override val name: String = "SELECT"
 
     override def validateRequest(securable: Securable): Boolean = securable match {
-      case _: Table | _: SecurableFunction => true
+      case _: Table | _: SecurableFunction | AnyFile => true
       case _ => false
     }
 
     override def validateGrant(securable: Securable): Boolean = securable match {
-      case Catalog | _: Database | _: Table | _: SecurableFunction => true
+      case Catalog | _: Database | _: Table | _: SecurableFunction | AnyFile => true
       case _ => false
     }
   }
@@ -290,12 +298,12 @@ object Action {
     override val name: String = "MODIFY"
 
     override def validateRequest(securable: Securable): Boolean = securable match {
-      case _: Table => true
+      case _: Table | AnyFile => true
       case _ => false
     }
 
     override def validateGrant(securable: Securable): Boolean = securable match {
-      case Catalog | _: Database | _: Table => true
+      case Catalog | _: Database | _: Table | AnyFile => true
       case _ => false
     }
   }

sql/core/src/test/scala/com/databricks/sql/acl/AclCommandParseSuite.scala

@@ -81,6 +81,12 @@ class AclCommandParseSuite extends PlanTest {
         client,
         Permission(user3, Select, AnonymousFunction) :: Nil))
 
+    checkAnswer(
+      "GRANT SELECT ON ANY FILE TO user3",
+      GrantPermissionsCommand(
+        client,
+        Permission(user3, Select, AnyFile) :: Nil))
+
     intercept("GRANT BLOB ON x TO `r2d2`", "Unknown ActionType")
 
     intercept("GRANT SELECT ON CAR x TO `r2d2`")
@@ -116,6 +122,12 @@ class AclCommandParseSuite extends PlanTest {
         client,
         Permission(user3, Select, AnonymousFunction) :: Nil))
 
+    checkAnswer(
+      "REVOKE SELECT ON ANY FILE FROM user3",
+      RevokePermissionsCommand(
+        client,
+        Permission(user3, Select, AnyFile) :: Nil))
+
     intercept("GRANT MOVE ON x TO c3po", "Unknown ActionType")
 
     intercept("GRANT SELECT ON TAXI x TO c3po")

sql/core/src/test/scala/com/databricks/sql/acl/CheckPermissionRuleSuite.scala

@@ -8,7 +8,7 @@
  */
 package com.databricks.sql.acl
 
-import com.databricks.sql.acl.Action.ReadMetadata
+import com.databricks.sql.acl.Action.{ReadMetadata, Select}
 import org.apache.hadoop.fs.Path
 import org.mockito.Mockito._
 
@@ -98,6 +98,12 @@ class CheckPermissionRuleSuite extends CheckRequests {
     LogicalRelation(mockBaseRel, None, Some(createCatalogTable(temp1)))
   }
 
+  protected def anonymousRel(): LogicalRelation = {
+    val mockBaseRel = mock(classOf[HadoopFsRelation])
+    when(mockBaseRel.schema).thenReturn(new StructType)
+    LogicalRelation(mockBaseRel, None, None)
+  }
+
   val db: String = "PERM1"
   val V1: TableIdentifier = TableIdentifier("V1", Some(db))
   val T1: TableIdentifier = TableIdentifier("T1", Some(db))
@@ -193,7 +199,8 @@ class CheckPermissionRuleSuite extends CheckRequests {
 
   private val ddlOnlyCommands: Seq[(String, LogicalPlan, Set[Request])] = Seq(
     addDDL("create temp view", CreateTempViewUsing(
-      TableIdentifier("blah"), None, false, false, null, Map.empty)),
+      TableIdentifier("blah"), None, false, false, null, Map.empty),
+      Request(AnyFile, Action.Select)),
     addDDL("createdstable", CreateDataSourceTableCommand(
       createCatalogTable(T1), false),
       Request(Database(T1.database.get), Action.Create)),
@@ -231,7 +238,16 @@ class CheckPermissionRuleSuite extends CheckRequests {
     addDDL("dropfunc", DropFunctionCommand(func1.database, func1.funcName, false, false),
       Request(Function(func1), Action.Own)),
     addDDL("dropfunctmp", DropFunctionCommand(tempFunc1.database, tempFunc1.funcName, false, true)),
-    addDDL("noop", NoOpRunnableCommand)
+    addDDL("noop", NoOpRunnableCommand),
+    addDDL("showdatabases", ShowDatabasesCommand(None), ReadMetadata(Catalog)),
+    addDDL("showtables", ShowTablesCommand(Option(db), None), ReadMetadata(Database(db))),
+    addDDL("showfunctions", ShowFunctionsCommand(Option(db), None, true, true),
+      ReadMetadata(Database(db))),
+    addDDL("showpartitions", ShowPartitionsCommand(T1, None), ReadMetadata(Table(T1))),
+    addDDL("showcolmns", ShowColumnsCommand(None, T1), ReadMetadata(Table(T1))),
+    addDDL("showcreatetable", ShowCreateTableCommand(T1), ReadMetadata(Table(T1))),
+    addDDL("showpermissions", ShowPermissionsCommand(NoOpAclClient, None, Catalog)),
+    addDDL("logicalrelation", anonymousRel(), Select(AnyFile))
   )
   private def mockedLogicalRelation(tableId: TableIdentifier): LogicalRelation = {
     val mocked = mock(classOf[LogicalRelation])

sql/core/src/test/scala/com/databricks/sql/acl/StringSerializationSuite.scala

@@ -105,6 +105,11 @@ class StringSerializationSuite extends SparkFunSuite{
     assert(e.getMessage.contains("A database is not defined for FUNCTION"))
   }
 
+  test("anonymous objects") {
+    roundTripSecurable(AnyFile, "/ANY_FILE")
+    roundTripSecurable(AnonymousFunction, "/ANONYMOUS_FUNCTION")
+  }
+
   test("invalid names") {
     def checkFail(in: String): Unit = {
       val e = intercept[IllegalArgumentException](Securable.fromString(in))

sql/hive/src/test/scala/com/databricks/sql/acl/HiveSQLCheckPermissionRuleSuite.scala

@@ -8,6 +8,8 @@
  */
 package com.databricks.sql.acl
 
+import java.nio.file.Files
+
 import org.scalatest.BeforeAndAfterEach
 
 import org.apache.spark.sql._
@@ -539,4 +541,28 @@ class HiveSQLCheckPermissionRuleSuite extends TestHiveExtensions(TestHiveAclExte
     assertFail("drop function perm.fnx")
     assertNoOp("drop function if exists perm.fnx")
   }
+
+  test("query on file") {
+    val path = Files.createTempDirectory("somedata").toString + "/SC-5483"
+    asUser("super") {
+      spark.range(1000).write.parquet(path)
+    }
+    asUser("U") {
+      intercept[SecurityException] {
+        spark.sql(s"select * from parquet.`$path`")
+      }
+      intercept[SecurityException] {
+        spark.sql(s"create temporary view x using parquet options(path '$path')")
+      }
+    }
+    asUser("super") {
+      spark.sql("grant select on any file to U")
+    }
+    asUser("U") {
+      assert(spark.sql(s"select * from parquet.`$path`").count === 1000L)
+      spark.sql(s"create temporary view x using parquet options(path '$path')")
+      assert(spark.sql("select * from x").count === 1000L)
+      spark.sql(s"drop view x")
+    }
+  }
 }