[CARMEL-3154] authorization (#4)

* [CARMEL-3154] authorization * fix code style * fix code style * fix code style * fix ut * fix ut * ignore some ut
apache · Aug 10, 2020 · 171868b · 171868b
1 parent 81d384a
commit 171868b
Show file tree

Hide file tree

Showing 33 changed files with 5,021 additions and 42 deletions.
diff --git a/sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4 b/sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4
@@ -220,23 +220,81 @@ statement
     | TRUNCATE TABLE multipartIdentifier partitionSpec?                #truncateTable
     | MSCK REPAIR TABLE multipartIdentifier                            #repairTable
     | op=(ADD | LIST) identifier (STRING | .*?)                        #manageResource
-    | SET ROLE .*?                                                     #failNativeCommand
+    | SET ROLE (ALL|identifier)                                        #setRole
     | SET .*?                                                          #setConfiguration
     | RESET                                                            #resetConfiguration
+    | CREATE ROLE identifier                                           #createRole
+    | DROP ROLE identifier                                             #dropRole
+    | GRANT privilegeList ON privilegeObject TO
+        principalSpecification withGrantOption?                        #grantPrivilege
+    | REVOKE grantOptionFor? privilegeList ON privilegeObject FROM
+        principalSpecification                                         #revokePrivilege
+    | GRANT ROLE? identifier (',' identifier)* TO
+        principalSpecification withAdminOption?                        #grantRole
+    | REVOKE adminOptionFor? ROLE? identifier (',' identifier)*
+        FROM principalSpecification                                    #revokeRole
+    | SHOW ROLE GRANT principalName                                    #showRoleGrants
+    | SHOW ROLES                                                       #showRoles
+    | SHOW CURRENT ROLES                                               #showCurrentRoles
+    | SHOW GRANT principalName? ON privilegeShowObject                 #showGrants
+    | SHOW PRINCIPALS identifier                                       #showRolePrincipals
     | unsupportedHiveNativeCommands .*?                                #failNativeCommand
     ;
 
+
+privilegeList
+    : privilegeDef (',' privilegeDef)*
+    ;
+
+privilegeDef
+    : privilegeType
+    ;
+
+privilegeType
+    : ALL
+    | SELECT
+    | INSERT
+    | UPDATE
+    | DELETE
+    ;
+
+privilegeObject
+    : DATABASE identifier
+    | TABLE? tableIdentifier
+    ;
+
+privilegeShowObject
+    : ALL
+    | privilegeObject
+    ;
+
+principalSpecification
+    : principalName (',' principalName)*
+    ;
+
+principalName
+    : USER identifier
+    | ROLE identifier
+    ;
+
+withGrantOption
+    : WITH GRANT OPTION
+    ;
+
+grantOptionFor
+    : GRANT OPTION FOR
+    ;
+
+withAdminOption
+    : WITH ADMIN OPTION
+    ;
+
+adminOptionFor
+    : ADMIN OPTION FOR
+    ;
+
 unsupportedHiveNativeCommands
-    : kw1=CREATE kw2=ROLE
-    | kw1=DROP kw2=ROLE
-    | kw1=GRANT kw2=ROLE?
-    | kw1=REVOKE kw2=ROLE?
-    | kw1=SHOW kw2=GRANT
-    | kw1=SHOW kw2=ROLE kw3=GRANT?
-    | kw1=SHOW kw2=PRINCIPALS
-    | kw1=SHOW kw2=ROLES
-    | kw1=SHOW kw2=CURRENT kw3=ROLES
-    | kw1=EXPORT kw2=TABLE
+    : kw1=EXPORT kw2=TABLE
     | kw1=IMPORT kw2=TABLE
     | kw1=SHOW kw2=COMPACTIONS
     | kw1=SHOW kw2=CREATE kw3=TABLE
@@ -991,6 +1049,7 @@ alterColumnAction
 // The non-reserved keywords are listed below. Keywords not in this list are reserved keywords.
 ansiNonReserved
     : ADD
+    | ADMIN
     | AFTER
     | ALTER
     | ANALYZE
@@ -1025,6 +1084,7 @@ ansiNonReserved
     | DATABASES
     | DBPROPERTIES
     | DEFINED
+    | DENY
     | DELETE
     | DELIMITED
     | DESC
@@ -1196,6 +1256,7 @@ strictNonReserved
 
 nonReserved
     : ADD
+    | ADMIN
     | AFTER
     | ALL
     | ALTER
@@ -1250,6 +1311,7 @@ nonReserved
     | DEFINED
     | DELETE
     | DELIMITED
+    | DENY
     | DESC
     | DESCRIBE
     | DFS
@@ -1443,6 +1505,7 @@ nonReserved
 // Start of the keywords list
 //============================
 ADD: 'ADD';
+ADMIN: 'ADMIN';
 AFTER: 'AFTER';
 ALL: 'ALL';
 ALTER: 'ALTER';
@@ -1499,6 +1562,7 @@ DBPROPERTIES: 'DBPROPERTIES';
 DEFINED: 'DEFINED';
 DELETE: 'DELETE';
 DELIMITED: 'DELIMITED';
+DENY: 'DENY';
 DESC: 'DESC';
 DESCRIBE: 'DESCRIBE';
 DFS: 'DFS';

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/QueryPlanningTracker.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/QueryPlanningTracker.scala
@@ -38,6 +38,7 @@ object QueryPlanningTracker {
   // Define a list of common phases here.
   val PARSING = "parsing"
   val ANALYSIS = "analysis"
+  val AUTHORIZE = "authorize"
   val OPTIMIZATION = "optimization"
   val PLANNING = "planning"
 

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
@@ -113,6 +113,8 @@ class SessionCatalog(
   @GuardedBy("this")
   protected var currentDb: String = formatDatabaseName(DEFAULT_DATABASE)
 
+  protected var currentUser: String = ""
+
   private val validNameFormat = "([\\w_]+)".r
 
   /**
@@ -278,6 +280,10 @@ class SessionCatalog(
     synchronized { currentDb = dbName }
   }
 
+  def setCurrentUser(username: String): Unit = {
+    synchronized { currentUser = username }
+  }
+
   /**
    * Get the path for creating a non-default database when database location is not provided
    * by users.
@@ -333,7 +339,7 @@ class SessionCatalog(
     } else if (validateLocation) {
       validateTableLocation(newTableDefinition)
     }
-    externalCatalog.createTable(newTableDefinition, ignoreIfExists)
+    externalCatalog.createTable(newTableDefinition.copy(owner = currentUser), ignoreIfExists)
   }
 
   def validateTableLocation(table: CatalogTable): Unit = {

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/interface.scala
@@ -28,6 +28,10 @@ import org.apache.spark.internal.Logging
 import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.{FunctionIdentifier, InternalRow, TableIdentifier}
 import org.apache.spark.sql.catalyst.analysis.MultiInstanceRelation
+import org.apache.spark.sql.catalyst.catalog.PrincipalType.PrincipalType
+import org.apache.spark.sql.catalyst.catalog.PrivilegeObjectActionType.PrivilegeObjectActionType
+import org.apache.spark.sql.catalyst.catalog.PrivilegeObjectType.PrivilegeObjectType
+import org.apache.spark.sql.catalyst.catalog.PrivilegeType.PrivilegeType
 import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap, AttributeReference, Cast, ExprId, Literal}
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.plans.logical.statsEstimation.EstimationUtils
@@ -37,6 +41,58 @@ import org.apache.spark.sql.connector.catalog.CatalogManager
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
 
+case class CatalogRole(name: String,
+    createTime: Long,
+    owner: String)
+
+case class CatalogPrincipal(name: String, principleType: PrincipalType) {
+  override def toString: String = s"Principal [name=$name, type=${principleType.toString}]"
+}
+
+object PrincipalType extends Enumeration {
+  type PrincipalType = Value
+  val USER, ROLE, GROUP = Value
+}
+
+case class CatalogPrivilege(privilege: PrivilegeType)
+
+case class CatalogPrivilegeObject(objectType: PrivilegeObjectType,
+    db: String, objectName: String,
+    actionType: PrivilegeObjectActionType = PrivilegeObjectActionType.OTHER) {
+  override def toString: String = if (objectType == PrivilegeObjectType.DATABASE) {
+    s"Object [type=${objectType.toString}, name=${objectName}]"
+  } else {
+    s"Object [type=${objectType.toString}, name=${db}.${objectName}]"
+  }
+}
+
+object PrivilegeObjectActionType extends Enumeration {
+  type PrivilegeObjectActionType = Value
+  val OTHER, INSERT, INSERT_OVERWRITE, UPDATE, DELETE = Value
+}
+
+object PrivilegeType extends Enumeration {
+  type PrivilegeType = Value
+  val ALL, SELECT, INSERT, UPDATE, DELETE = Value
+}
+
+object PrivilegeObjectType extends Enumeration {
+  type PrivilegeObjectType = Value
+  val GLOBAL, DATABASE, TABLE, PARTITION, COLUMN = Value
+}
+
+case class CatalogRoleGrant(roleName: String,
+    principal: CatalogPrincipal,
+    grantOption: Boolean,
+    grantTime: Long,
+    grantor: CatalogPrincipal)
+
+case class CatalogPrivilegeGrant(privilegeObject: CatalogPrivilegeObject,
+    principal: CatalogPrincipal,
+    privilege: CatalogPrivilege,
+    grantOption: Boolean,
+    grantTime: Long,
+    grantor: CatalogPrincipal)
 
 /**
  * A function defined in the catalog.

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
@@ -30,7 +30,8 @@ import org.apache.spark.internal.Logging
 import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.{FunctionIdentifier, TableIdentifier}
 import org.apache.spark.sql.catalyst.analysis._
-import org.apache.spark.sql.catalyst.catalog.{BucketSpec, CatalogStorageFormat, CatalogUtils, FunctionResource, FunctionResourceType}
+import org.apache.spark.sql.catalyst.catalog._
+import org.apache.spark.sql.catalyst.catalog.PrivilegeType.PrivilegeType
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.expressions.aggregate.{First, Last}
 import org.apache.spark.sql.catalyst.parser.SqlBaseParser._
@@ -3660,4 +3661,62 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
     val nameParts = visitMultipartIdentifier(ctx.multipartIdentifier)
     CommentOnTable(UnresolvedTable(nameParts), comment)
   }
+
+  override def visitPrivilegeList(
+      ctx: PrivilegeListContext): Seq[CatalogPrivilege] = withOrigin(ctx) {
+    ctx.privilegeDef().asScala.map(visitPrivilegeDef)
+  }
+
+  override def visitPrivilegeDef(ctx: PrivilegeDefContext): CatalogPrivilege = withOrigin(ctx) {
+    CatalogPrivilege(visitPrivilegeType(ctx.privilegeType))
+  }
+
+  override def visitPrivilegeType(ctx: PrivilegeTypeContext): PrivilegeType = withOrigin(ctx) {
+    if (ctx.ALL() != null) {
+      PrivilegeType.ALL
+    } else if (ctx.SELECT() != null) {
+      PrivilegeType.SELECT
+    } else if (ctx.INSERT() != null) {
+      PrivilegeType.INSERT
+    } else if (ctx.UPDATE() != null) {
+      PrivilegeType.UPDATE
+    } else if (ctx.DELETE() != null) {
+      PrivilegeType.DELETE
+    } else {
+      throw new ParseException("Privilege type is not supported", ctx)
+    }
+  }
+
+  override def visitPrivilegeShowObject(
+      ctx: PrivilegeShowObjectContext): Option[CatalogPrivilegeObject] = withOrigin(ctx) {
+    if (ctx.ALL() != null) {
+      None
+    } else {
+      Some(visitPrivilegeObject(ctx.privilegeObject()))
+    }
+  }
+  override def visitPrivilegeObject(ctx: PrivilegeObjectContext): CatalogPrivilegeObject =
+    withOrigin(ctx) {
+      if (ctx.DATABASE() != null) {
+        CatalogPrivilegeObject(PrivilegeObjectType.DATABASE,
+          ctx.identifier.getText, ctx.identifier.getText)
+      } else {
+        val tableIdentifier = visitTableIdentifier(ctx.tableIdentifier())
+        CatalogPrivilegeObject(PrivilegeObjectType.TABLE,
+          tableIdentifier.database.getOrElse(""), tableIdentifier.table)
+      }
+    }
+
+  override def visitPrincipalSpecification(
+      ctx: PrincipalSpecificationContext): Seq[CatalogPrincipal] = withOrigin(ctx) {
+    ctx.principalName.asScala.map(visitPrincipalName)
+  }
+
+  override def visitPrincipalName(ctx: PrincipalNameContext): CatalogPrincipal = withOrigin(ctx) {
+    if (ctx.USER() != null) {
+      CatalogPrincipal(ctx.identifier.getText, PrincipalType.USER)
+    } else {
+      CatalogPrincipal(ctx.identifier.getText, PrincipalType.ROLE)
+    }
+  }
 }
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statements.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statements.scala
@@ -18,7 +18,7 @@
 package org.apache.spark.sql.catalyst.plans.logical
 
 import org.apache.spark.sql.catalyst.analysis.ViewType
-import org.apache.spark.sql.catalyst.catalog.{BucketSpec, FunctionResource}
+import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
 import org.apache.spark.sql.catalyst.expressions.Attribute
 import org.apache.spark.sql.connector.catalog.TableChange.ColumnPosition
@@ -476,3 +476,38 @@ case class CreateFunctionStatement(
     isTemp: Boolean,
     ignoreIfExists: Boolean,
     replace: Boolean) extends ParsedStatement
+
+case class GrantPrivilegeStatement(privileges: Seq[CatalogPrivilege],
+    privilegeObject: CatalogPrivilegeObject,
+    principals: Seq[CatalogPrincipal],
+    withGrant: Boolean) extends ParsedStatement
+
+case class RevokePrivilegeStatement(privileges: Seq[CatalogPrivilege],
+    privilegeObject: CatalogPrivilegeObject,
+    principals: Seq[CatalogPrincipal],
+    withGrant: Boolean) extends ParsedStatement
+
+case class GrantRoleStatement(principals: Seq[CatalogPrincipal],
+    roles: Seq[String],
+    withAdmin: Boolean) extends ParsedStatement
+
+case class RevokeRoleStatement(principals: Seq[CatalogPrincipal],
+    roles: Seq[String],
+    withAdmin: Boolean) extends ParsedStatement
+
+case class CreateRoleStatement(roleName: String) extends ParsedStatement
+
+case class DropRoleStatement(roleName: String) extends ParsedStatement
+
+case class ShowRoleGrantsStatement(principal: CatalogPrincipal) extends ParsedStatement
+
+case class ShowRolesStatement() extends ParsedStatement
+
+case class SetRoleStatement(roleName: String) extends ParsedStatement
+
+case class ShowCurrentRolesStatement() extends ParsedStatement
+
+case class ShowRolePrincipalsStatement(roleName: String) extends ParsedStatement
+
+case class ShowGrantsStatement(principal: Option[CatalogPrincipal],
+    privilege: Option[CatalogPrivilegeObject]) extends ParsedStatement
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/StaticSQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/StaticSQLConf.scala
@@ -54,6 +54,11 @@ object StaticSQLConf {
     .transform(_.toLowerCase(Locale.ROOT))
     .createWithDefault("global_temp")
 
+  val AUTHORIZATION_ENABLED = buildStaticConf("spark.sql.authorization.enabled")
+    .internal()
+    .booleanConf
+    .createWithDefault(false)
+
   // This is used to control when we will split a schema's JSON string to multiple pieces
   // in order to fit the JSON string in metastore's table property (by default, the value has
   // a length restriction of 4000 characters, so do not use a value larger than 4000 as the default