[SPARK-4224][CORE][YARN] Support group acls

## What changes were proposed in this pull request?
Currently only a list of users can be specified for view and modify acls. This change enables a group of admins/devs/users to be provisioned for viewing and modifying Spark jobs.

**Changes Proposed in the fix**
Three new corresponding config entries have been added where the user can specify the groups to be given access.

```
spark.admin.acls.groups
spark.modify.acls.groups
spark.ui.view.acls.groups
```

New config entries were added because specifying the users and groups explicitly is a better and cleaner way compared to specifying them in the existing config entry using a delimiter.

A generic trait has been introduced to provide the user to group mapping which makes it pluggable to support a variety of mapping protocols - similar to the one used in hadoop. A default unix shell based implementation has been provided.
Custom user to group mapping protocol can be specified and configured by the entry ```spark.user.groups.mapping```

**How the patch was Tested**
We ran different spark jobs setting the config entries in combinations of admin, modify and ui acls. For modify acls we tried killing the job stages from the ui and using yarn commands. For view acls we tried accessing the UI tabs and the logs. Headless accounts were used to launch these jobs and different users tried to modify and view the jobs to ensure that the groups mapping applied correctly.

Additional Unit tests have been added without modifying the existing ones. These test for different ways of setting the acls through configuration and/or API and validate the expected behavior.

Author: Dhruve Ashar <dhruveashar@gmail.com>

Closes #12760 from dhruve/impr/SPARK-4224.

7 files changed, 405 insertions, 25 deletions

diff --git a/core/src/main/scala/org/apache/spark/SecurityManager.scala b/core/src/main/scala/org/apache/spark/SecurityManager.scala
index e8f68224d5..f72c7ded5e 100644
--- a/core/src/main/scala/org/apache/spark/SecurityManager.scala
+++ b/core/src/main/scala/org/apache/spark/SecurityManager.scala
@@ -50,17 +50,19 @@ import org.apache.spark.util.Utils
  * secure the UI if it has data that other users should not be allowed to see. The javax
  * servlet filter specified by the user can authenticate the user and then once the user
  * is logged in, Spark can compare that user versus the view acls to make sure they are
- * authorized to view the UI. The configs 'spark.acls.enable' and 'spark.ui.view.acls'
- * control the behavior of the acls. Note that the person who started the application
- * always has view access to the UI.
+ * authorized to view the UI. The configs 'spark.acls.enable', 'spark.ui.view.acls' and
+ * 'spark.ui.view.acls.groups' control the behavior of the acls. Note that the person who
+ * started the application always has view access to the UI.
  *
- * Spark has a set of modify acls (`spark.modify.acls`) that controls which users have permission
- * to  modify a single application. This would include things like killing the application. By
- * default the person who started the application has modify access. For modify access through
- * the UI, you must have a filter that does authentication in place for the modify acls to work
- * properly.
+ * Spark has a set of individual and group modify acls (`spark.modify.acls`) and
+ * (`spark.modify.acls.groups`) that controls which users and groups have permission to
+ * modify a single application. This would include things like killing the application.
+ * By default the person who started the application has modify access. For modify access
+ * through the UI, you must have a filter that does authentication in place for the modify
+ * acls to work properly.
  *
- * Spark also has a set of admin acls (`spark.admin.acls`) which is a set of users/administrators
+ * Spark also has a set of individual and group admin acls (`spark.admin.acls`) and
+ * (`spark.admin.acls.groups`) which is a set of users/administrators and admin groups
  * who always have permission to view or modify the Spark application.
  *
  * Starting from version 1.3, Spark has partial support for encrypted connections with SSL.
@@ -184,6 +186,9 @@ private[spark] class SecurityManager(sparkConf: SparkConf)
 
   import SecurityManager._
 
+  // allow all users/groups to have view/modify permissions
+  private val WILDCARD_ACL = "*"
+
   private val authOn = sparkConf.getBoolean(SecurityManager.SPARK_AUTH_CONF, false)
   // keep spark.ui.acls.enable for backwards compatibility with 1.0
   private var aclsOn =
@@ -193,12 +198,20 @@ private[spark] class SecurityManager(sparkConf: SparkConf)
   private var adminAcls: Set[String] =
     stringToSet(sparkConf.get("spark.admin.acls", ""))
 
+  // admin group acls should be set before view or modify group acls
+  private var adminAclsGroups : Set[String] =
+    stringToSet(sparkConf.get("spark.admin.acls.groups", ""))
+
   private var viewAcls: Set[String] = _
 
+  private var viewAclsGroups: Set[String] = _
+
   // list of users who have permission to modify the application. This should
   // apply to both UI and CLI for things like killing the application.
   private var modifyAcls: Set[String] = _
 
+  private var modifyAclsGroups: Set[String] = _
+
   // always add the current user and SPARK_USER to the viewAcls
   private val defaultAclUsers = Set[String](System.getProperty("user.name", ""),
     Utils.getCurrentUserName())
@@ -206,11 +219,16 @@ private[spark] class SecurityManager(sparkConf: SparkConf)
   setViewAcls(defaultAclUsers, sparkConf.get("spark.ui.view.acls", ""))
   setModifyAcls(defaultAclUsers, sparkConf.get("spark.modify.acls", ""))
 
+  setViewAclsGroups(sparkConf.get("spark.ui.view.acls.groups", ""));
+  setModifyAclsGroups(sparkConf.get("spark.modify.acls.groups", ""));
+
   private val secretKey = generateSecretKey()
   logInfo("SecurityManager: authentication " + (if (authOn) "enabled" else "disabled") +
     "; ui acls " + (if (aclsOn) "enabled" else "disabled") +
-    "; users with view permissions: " + viewAcls.toString() +
-    "; users with modify permissions: " + modifyAcls.toString())
+    "; users  with view permissions: " + viewAcls.toString() +
+    "; groups with view permissions: " + viewAclsGroups.toString() +
+    "; users  with modify permissions: " + modifyAcls.toString() +
+    "; groups with modify permissions: " + modifyAclsGroups.toString())
 
   // Set our own authenticator to properly negotiate user/password for HTTP connections.
   // This is needed by the HTTP client fetching from the HttpServer. Put here so its
@@ -303,16 +321,33 @@ private[spark] class SecurityManager(sparkConf: SparkConf)
   }
 
   /**
+   * Admin acls groups should be set before the view or modify acls groups. If you modify the admin
+   * acls groups you should also set the view and modify acls groups again to pick up the changes.
+   */
+  def setViewAclsGroups(allowedUserGroups: String) {
+    viewAclsGroups = (adminAclsGroups ++ stringToSet(allowedUserGroups));
+    logInfo("Changing view acls groups to: " + viewAclsGroups.mkString(","))
+  }
+
+  /**
    * Checking the existence of "*" is necessary as YARN can't recognize the "*" in "defaultuser,*"
    */
   def getViewAcls: String = {
-    if (viewAcls.contains("*")) {
-      "*"
+    if (viewAcls.contains(WILDCARD_ACL)) {
+      WILDCARD_ACL
     } else {
       viewAcls.mkString(",")
     }
   }
 
+  def getViewAclsGroups: String = {
+    if (viewAclsGroups.contains(WILDCARD_ACL)) {
+      WILDCARD_ACL
+    } else {
+      viewAclsGroups.mkString(",")
+    }
+  }
+
   /**
    * Admin acls should be set before the view or modify acls.  If you modify the admin
    * acls you should also set the view and modify acls again to pick up the changes.
@@ -323,16 +358,33 @@ private[spark] class SecurityManager(sparkConf: SparkConf)
   }
 
   /**
+   * Admin acls groups should be set before the view or modify acls groups. If you modify the admin
+   * acls groups you should also set the view and modify acls groups again to pick up the changes.
+   */
+  def setModifyAclsGroups(allowedUserGroups: String) {
+    modifyAclsGroups = (adminAclsGroups ++ stringToSet(allowedUserGroups));
+    logInfo("Changing modify acls groups to: " + modifyAclsGroups.mkString(","))
+  }
+
+  /**
    * Checking the existence of "*" is necessary as YARN can't recognize the "*" in "defaultuser,*"
    */
   def getModifyAcls: String = {
-    if (modifyAcls.contains("*")) {
-      "*"
+    if (modifyAcls.contains(WILDCARD_ACL)) {
+      WILDCARD_ACL
     } else {
       modifyAcls.mkString(",")
     }
   }
 
+  def getModifyAclsGroups: String = {
+    if (modifyAclsGroups.contains(WILDCARD_ACL)) {
+      WILDCARD_ACL
+    } else {
+      modifyAclsGroups.mkString(",")
+    }
+  }
+
   /**
    * Admin acls should be set before the view or modify acls.  If you modify the admin
    * acls you should also set the view and modify acls again to pick up the changes.
@@ -342,6 +394,15 @@ private[spark] class SecurityManager(sparkConf: SparkConf)
     logInfo("Changing admin acls to: " + adminAcls.mkString(","))
   }
 
+  /**
+   * Admin acls groups should be set before the view or modify acls groups. If you modify the admin
+   * acls groups you should also set the view and modify acls groups again to pick up the changes.
+   */
+  def setAdminAclsGroups(adminUserGroups: String) {
+    adminAclsGroups = stringToSet(adminUserGroups)
+    logInfo("Changing admin acls groups to: " + adminAclsGroups.mkString(","))
+  }
+
   def setAcls(aclSetting: Boolean) {
     aclsOn = aclSetting
     logInfo("Changing acls enabled to: " + aclsOn)
@@ -398,36 +459,49 @@ private[spark] class SecurityManager(sparkConf: SparkConf)
   def aclsEnabled(): Boolean = aclsOn
 
   /**
-   * Checks the given user against the view acl list to see if they have
+   * Checks the given user against the view acl and groups list to see if they have
    * authorization to view the UI. If the UI acls are disabled
    * via spark.acls.enable, all users have view access. If the user is null
-   * it is assumed authentication is off and all users have access.
+   * it is assumed authentication is off and all users have access. Also if any one of the
+   * UI acls or groups specify the WILDCARD(*) then all users have view access.
    *
    * @param user to see if is authorized
    * @return true is the user has permission, otherwise false
    */
   def checkUIViewPermissions(user: String): Boolean = {
     logDebug("user=" + user + " aclsEnabled=" + aclsEnabled() + " viewAcls=" +
-      viewAcls.mkString(","))
-    !aclsEnabled || user == null || viewAcls.contains(user) || viewAcls.contains("*")
+      viewAcls.mkString(",") + " viewAclsGroups=" + viewAclsGroups.mkString(","))
+    if (!aclsEnabled || user == null || viewAcls.contains(user) ||
+        viewAcls.contains(WILDCARD_ACL) || viewAclsGroups.contains(WILDCARD_ACL)) {
+      return true
+    }
+    val currentUserGroups = Utils.getCurrentUserGroups(sparkConf, user)
+    logDebug("userGroups=" + currentUserGroups.mkString(","))
+    viewAclsGroups.exists(currentUserGroups.contains(_))
   }
 
   /**
-   * Checks the given user against the modify acl list to see if they have
-   * authorization to modify the application. If the UI acls are disabled
+   * Checks the given user against the modify acl and groups list to see if they have
+   * authorization to modify the application. If the modify acls are disabled
    * via spark.acls.enable, all users have modify access. If the user is null
-   * it is assumed authentication isn't turned on and all users have access.
+   * it is assumed authentication isn't turned on and all users have access. Also if any one
+   * of the modify acls or groups specify the WILDCARD(*) then all users have modify access.
    *
    * @param user to see if is authorized
    * @return true is the user has permission, otherwise false
    */
   def checkModifyPermissions(user: String): Boolean = {
     logDebug("user=" + user + " aclsEnabled=" + aclsEnabled() + " modifyAcls=" +
-      modifyAcls.mkString(","))
-    !aclsEnabled || user == null || modifyAcls.contains(user) || modifyAcls.contains("*")
+      modifyAcls.mkString(",") + " modifyAclsGroups=" + modifyAclsGroups.mkString(","))
+    if (!aclsEnabled || user == null || modifyAcls.contains(user) ||
+        modifyAcls.contains(WILDCARD_ACL) || modifyAclsGroups.contains(WILDCARD_ACL)) {
+      return true
+    }
+    val currentUserGroups = Utils.getCurrentUserGroups(sparkConf, user)
+    logDebug("userGroups=" + currentUserGroups)
+    modifyAclsGroups.exists(currentUserGroups.contains(_))
   }
 
-
   /**
    * Check to see if authentication for the Spark communication protocols is enabled
    * @return true if authentication is enabled, otherwise false
diff --git a/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala b/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
index 07cbcec8e5..110d882f05 100644
--- a/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
@@ -245,6 +245,8 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
             ui.getSecurityManager.setAdminAcls(appListener.adminAcls.getOrElse(""))
             ui.getSecurityManager.setViewAcls(attempt.sparkUser,
               appListener.viewAcls.getOrElse(""))
+            ui.getSecurityManager.setAdminAclsGroups(appListener.adminAclsGroups.getOrElse(""))
+            ui.getSecurityManager.setViewAclsGroups(appListener.viewAclsGroups.getOrElse(""))
             LoadedAppUI(ui, updateProbe(appId, attemptId, attempt.fileSize))
           }
         }
diff --git a/core/src/main/scala/org/apache/spark/scheduler/ApplicationEventListener.scala b/core/src/main/scala/org/apache/spark/scheduler/ApplicationEventListener.scala
index 9f218c64ca..28c45d800e 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/ApplicationEventListener.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/ApplicationEventListener.scala
@@ -32,6 +32,8 @@ private[spark] class ApplicationEventListener extends SparkListener {
   var endTime: Option[Long] = None
   var viewAcls: Option[String] = None
   var adminAcls: Option[String] = None
+  var viewAclsGroups: Option[String] = None
+  var adminAclsGroups: Option[String] = None
 
   override def onApplicationStart(applicationStart: SparkListenerApplicationStart) {
     appName = Some(applicationStart.appName)
@@ -51,6 +53,8 @@ private[spark] class ApplicationEventListener extends SparkListener {
       val allProperties = environmentDetails("Spark Properties").toMap
       viewAcls = allProperties.get("spark.ui.view.acls")
       adminAcls = allProperties.get("spark.admin.acls")
+      viewAclsGroups = allProperties.get("spark.ui.view.acls.groups")
+      adminAclsGroups = allProperties.get("spark.admin.acls.groups")
     }
   }
 }
diff --git a/core/src/main/scala/org/apache/spark/security/GroupMappingServiceProvider.scala b/core/src/main/scala/org/apache/spark/security/GroupMappingServiceProvider.scala
new file mode 100644
index 0000000000..ea047a4f75
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/security/GroupMappingServiceProvider.scala
@@ -0,0 +1,38 @@
+/*
+ * 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.spark.security
+
+/**
+ * This Spark trait is used for mapping a given userName to a set of groups which it belongs to.
+ * This is useful for specifying a common group of admins/developers to provide them admin, modify
+ * and/or view access rights. Based on whether access control checks are enabled using
+ * spark.acls.enable, every time a user tries to access or modify the application, the
+ * SecurityManager gets the corresponding groups a user belongs to from the instance of the groups
+ * mapping provider specified by the entry spark.user.groups.mapping.
+ */
+
+trait GroupMappingServiceProvider {
+
+  /**
+   * Get the groups the user belongs to.
+   * @param userName User's Name
+   * @return set of groups that the user belongs to. Empty in case of an invalid user.
+   */
+  def getGroups(userName : String) : Set[String]
+
+}
diff --git a/core/src/main/scala/org/apache/spark/security/ShellBasedGroupsMappingProvider.scala b/core/src/main/scala/org/apache/spark/security/ShellBasedGroupsMappingProvider.scala
new file mode 100644
index 0000000000..f71dd08246
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/security/ShellBasedGroupsMappingProvider.scala
@@ -0,0 +1,45 @@
+/*
+ * 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.spark.security
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.util.Utils
+
+/**
+ * This class is responsible for getting the groups for a particular user in Unix based
+ * environments. This implementation uses the Unix Shell based id command to fetch the user groups
+ * for the specified user. It does not cache the user groups as the invocations are expected
+ * to be infrequent.
+ */
+
+private[spark] class ShellBasedGroupsMappingProvider extends GroupMappingServiceProvider
+  with Logging {
+
+  override def getGroups(username: String): Set[String] = {
+    val userGroups = getUnixGroups(username)
+    logDebug("User: " + username + " Groups: " + userGroups.mkString(","))
+    userGroups
+  }
+
+  // shells out a "bash -c id -Gn username" to get user groups
+  private def getUnixGroups(username: String): Set[String] = {
+    val cmdSeq = Seq("bash", "-c", "id -Gn " + username)
+    // we need to get rid of the trailing "\n" from the result of command execution
+    Utils.executeAndGetOutput(cmdSeq).stripLineEnd.split(" ").toSet
+  }
+}
diff --git a/core/src/main/scala/org/apache/spark/util/Utils.scala b/core/src/main/scala/org/apache/spark/util/Utils.scala
index ea49991493..a8bb0002a7 100644
--- a/core/src/main/scala/org/apache/spark/util/Utils.scala
+++ b/core/src/main/scala/org/apache/spark/util/Utils.scala
@@ -2181,6 +2181,25 @@ private[spark] object Utils extends Logging {
       .getOrElse(UserGroupInformation.getCurrentUser().getShortUserName())
   }
 
+  val EMPTY_USER_GROUPS = Set[String]()
+
+  // Returns the groups to which the current user belongs.
+  def getCurrentUserGroups(sparkConf: SparkConf, username: String): Set[String] = {
+    val groupProviderClassName = sparkConf.get("spark.user.groups.mapping",
+      "org.apache.spark.security.ShellBasedGroupsMappingProvider")
+    if (groupProviderClassName != "") {
+      try {
+        val groupMappingServiceProvider = classForName(groupProviderClassName).newInstance.
+          asInstanceOf[org.apache.spark.security.GroupMappingServiceProvider]
+        val currentUserGroups = groupMappingServiceProvider.getGroups(username)
+        return currentUserGroups
+      } catch {
+        case e: Exception => logError(s"Error getting groups for user=$username", e)
+      }
+    }
+    EMPTY_USER_GROUPS
+  }
+
   /**
    * Split the comma delimited string of master URLs into a list.
    * For instance, "spark://abc,def" becomes [spark://abc, spark://def].
diff --git a/core/src/test/scala/org/apache/spark/SecurityManagerSuite.scala b/core/src/test/scala/org/apache/spark/SecurityManagerSuite.scala
index 8bdb237c28..9801b2638c 100644
--- a/core/src/test/scala/org/apache/spark/SecurityManagerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/SecurityManagerSuite.scala
@@ -19,8 +19,18 @@ package org.apache.spark
 
 import java.io.File
 
+import org.apache.spark.security.GroupMappingServiceProvider
 import org.apache.spark.util.{ResetSystemProperties, SparkConfWithEnv, Utils}
 
+class DummyGroupMappingServiceProvider extends GroupMappingServiceProvider {
+
+  val userGroups: Set[String] = Set[String]("group1", "group2", "group3")
+
+  override def getGroups(username: String): Set[String] = {
+    userGroups
+  }
+}
+
 class SecurityManagerSuite extends SparkFunSuite with ResetSystemProperties {
 
   test("set security with conf") {
@@ -37,6 +47,45 @@ class SecurityManagerSuite extends SparkFunSuite with ResetSystemProperties {
     assert(securityManager.checkUIViewPermissions("user3") === false)
   }
 
+  test("set security with conf for groups") {
+    val conf = new SparkConf
+    conf.set("spark.authenticate", "true")
+    conf.set("spark.authenticate.secret", "good")
+    conf.set("spark.ui.acls.enable", "true")
+    conf.set("spark.ui.view.acls.groups", "group1,group2")
+    // default ShellBasedGroupsMappingProvider is used to resolve user groups
+    val securityManager = new SecurityManager(conf);
+    // assuming executing user does not belong to group1,group2
+    assert(securityManager.checkUIViewPermissions("user1") === false)
+    assert(securityManager.checkUIViewPermissions("user2") === false)
+
+    val conf2 = new SparkConf
+    conf2.set("spark.authenticate", "true")
+    conf2.set("spark.authenticate.secret", "good")
+    conf2.set("spark.ui.acls.enable", "true")
+    conf2.set("spark.ui.view.acls.groups", "group1,group2")
+    // explicitly specify a custom GroupsMappingServiceProvider
+    conf2.set("spark.user.groups.mapping", "org.apache.spark.DummyGroupMappingServiceProvider")
+
+    val securityManager2 = new SecurityManager(conf2);
+    // group4,group5 do not match
+    assert(securityManager2.checkUIViewPermissions("user1") === true)
+    assert(securityManager2.checkUIViewPermissions("user2") === true)
+
+    val conf3 = new SparkConf
+    conf3.set("spark.authenticate", "true")
+    conf3.set("spark.authenticate.secret", "good")
+    conf3.set("spark.ui.acls.enable", "true")
+    conf3.set("spark.ui.view.acls.groups", "group4,group5")
+    // explicitly specify a bogus GroupsMappingServiceProvider
+    conf3.set("spark.user.groups.mapping", "BogusServiceProvider")
+
+    val securityManager3 = new SecurityManager(conf3);
+    // BogusServiceProvider cannot be loaded and an error is logged returning an empty group set
+    assert(securityManager3.checkUIViewPermissions("user1") === false)
+    assert(securityManager3.checkUIViewPermissions("user2") === false)
+  }
+
   test("set security with api") {
     val conf = new SparkConf
     conf.set("spark.ui.view.acls", "user1,user2")
@@ -60,6 +109,40 @@ class SecurityManagerSuite extends SparkFunSuite with ResetSystemProperties {
     assert(securityManager.checkUIViewPermissions(null) === true)
   }
 
+  test("set security with api for groups") {
+    val conf = new SparkConf
+    conf.set("spark.user.groups.mapping", "org.apache.spark.DummyGroupMappingServiceProvider")
+
+    val securityManager = new SecurityManager(conf);
+    securityManager.setAcls(true)
+    securityManager.setViewAclsGroups("group1,group2")
+
+    // group1,group2 match
+    assert(securityManager.checkUIViewPermissions("user1") === true)
+    assert(securityManager.checkUIViewPermissions("user2") === true)
+
+    // change groups so they do not match
+    securityManager.setViewAclsGroups("group4,group5")
+    assert(securityManager.checkUIViewPermissions("user1") === false)
+    assert(securityManager.checkUIViewPermissions("user2") === false)
+
+    val conf2 = new SparkConf
+    conf.set("spark.user.groups.mapping", "BogusServiceProvider")
+
+    val securityManager2 = new SecurityManager(conf2)
+    securityManager2.setAcls(true)
+    securityManager2.setViewAclsGroups("group1,group2")
+
+    // group1,group2 do not match because of BogusServiceProvider
+    assert(securityManager.checkUIViewPermissions("user1") === false)
+    assert(securityManager.checkUIViewPermissions("user2") === false)
+
+    // setting viewAclsGroups to empty should still not match because of BogusServiceProvider
+    securityManager2.setViewAclsGroups("")
+    assert(securityManager.checkUIViewPermissions("user1") === false)
+    assert(securityManager.checkUIViewPermissions("user2") === false)
+  }
+
   test("set security modify acls") {
     val conf = new SparkConf
     conf.set("spark.modify.acls", "user1,user2")
@@ -84,6 +167,29 @@ class SecurityManagerSuite extends SparkFunSuite with ResetSystemProperties {
     assert(securityManager.checkModifyPermissions(null) === true)
   }
 
+  test("set security modify acls for groups") {
+    val conf = new SparkConf
+    conf.set("spark.user.groups.mapping", "org.apache.spark.DummyGroupMappingServiceProvider")
+
+    val securityManager = new SecurityManager(conf);
+    securityManager.setAcls(true)
+    securityManager.setModifyAclsGroups("group1,group2")
+
+    // group1,group2 match
+    assert(securityManager.checkModifyPermissions("user1") === true)
+    assert(securityManager.checkModifyPermissions("user2") === true)
+
+    // change groups so they do not match
+    securityManager.setModifyAclsGroups("group4,group5")
+    assert(securityManager.checkModifyPermissions("user1") === false)
+    assert(securityManager.checkModifyPermissions("user2") === false)
+
+    // change so they match again
+    securityManager.setModifyAclsGroups("group2,group3")
+    assert(securityManager.checkModifyPermissions("user1") === true)
+    assert(securityManager.checkModifyPermissions("user2") === true)
+  }
+
   test("set security admin acls") {
     val conf = new SparkConf
     conf.set("spark.admin.acls", "user1,user2")
@@ -122,7 +228,48 @@ class SecurityManagerSuite extends SparkFunSuite with ResetSystemProperties {
     assert(securityManager.checkUIViewPermissions("user1") === false)
     assert(securityManager.checkUIViewPermissions("user3") === false)
     assert(securityManager.checkUIViewPermissions(null) === true)
+  }
+
+  test("set security admin acls for groups") {
+    val conf = new SparkConf
+    conf.set("spark.admin.acls.groups", "group1")
+    conf.set("spark.ui.view.acls.groups", "group2")
+    conf.set("spark.modify.acls.groups", "group3")
+    conf.set("spark.user.groups.mapping", "org.apache.spark.DummyGroupMappingServiceProvider")
+
+    val securityManager = new SecurityManager(conf);
+    securityManager.setAcls(true)
+    assert(securityManager.aclsEnabled() === true)
+
+    // group1,group2,group3 match
+    assert(securityManager.checkModifyPermissions("user1") === true)
+    assert(securityManager.checkUIViewPermissions("user1") === true)
 
+    // change admin groups so they do not match. view and modify groups are set to admin groups
+    securityManager.setAdminAclsGroups("group4,group5")
+    // invoke the set ui and modify to propagate the changes
+    securityManager.setViewAclsGroups("")
+    securityManager.setModifyAclsGroups("")
+
+    assert(securityManager.checkModifyPermissions("user1") === false)
+    assert(securityManager.checkUIViewPermissions("user1") === false)
+
+    // change modify groups so they match
+    securityManager.setModifyAclsGroups("group3")
+    assert(securityManager.checkModifyPermissions("user1") === true)
+    assert(securityManager.checkUIViewPermissions("user1") === false)
+
+    // change view groups so they match
+    securityManager.setViewAclsGroups("group2")
+    securityManager.setModifyAclsGroups("group4")
+    assert(securityManager.checkModifyPermissions("user1") === false)
+    assert(securityManager.checkUIViewPermissions("user1") === true)
+
+    // change modify and view groups so they do not match
+    securityManager.setViewAclsGroups("group7")
+    securityManager.setModifyAclsGroups("group8")
+    assert(securityManager.checkModifyPermissions("user1") === false)
+    assert(securityManager.checkUIViewPermissions("user1") === false)
   }
 
   test("set security with * in acls") {
@@ -166,6 +313,57 @@ class SecurityManagerSuite extends SparkFunSuite with ResetSystemProperties {
     assert(securityManager.checkModifyPermissions("user8") === true)
   }
 
+  test("set security with * in acls for groups") {
+    val conf = new SparkConf
+    conf.set("spark.ui.acls.enable", "true")
+    conf.set("spark.admin.acls.groups", "group4,group5")
+    conf.set("spark.ui.view.acls.groups", "*")
+    conf.set("spark.modify.acls.groups", "group6")
+
+    val securityManager = new SecurityManager(conf)
+    assert(securityManager.aclsEnabled() === true)
+
+    // check for viewAclsGroups with *
+    assert(securityManager.checkUIViewPermissions("user1") === true)
+    assert(securityManager.checkUIViewPermissions("user2") === true)
+    assert(securityManager.checkModifyPermissions("user1") === false)
+    assert(securityManager.checkModifyPermissions("user2") === false)
+
+    // check for modifyAcls with *
+    securityManager.setModifyAclsGroups("*")
+    securityManager.setViewAclsGroups("group6")
+    assert(securityManager.checkUIViewPermissions("user1") === false)
+    assert(securityManager.checkUIViewPermissions("user2") === false)
+    assert(securityManager.checkModifyPermissions("user1") === true)
+    assert(securityManager.checkModifyPermissions("user2") === true)
+
+    // check for adminAcls with *
+    securityManager.setAdminAclsGroups("group9,*")
+    securityManager.setModifyAclsGroups("group4,group5")
+    securityManager.setViewAclsGroups("group6,group7")
+    assert(securityManager.checkUIViewPermissions("user5") === true)
+    assert(securityManager.checkUIViewPermissions("user6") === true)
+    assert(securityManager.checkModifyPermissions("user7") === true)
+    assert(securityManager.checkModifyPermissions("user8") === true)
+  }
+
+  test("security for groups default behavior") {
+    // no groups or userToGroupsMapper provided
+    // this will default to the ShellBasedGroupsMappingProvider
+    val conf = new SparkConf
+
+    val securityManager = new SecurityManager(conf)
+    securityManager.setAcls(true)
+
+    assert(securityManager.checkUIViewPermissions("user1") === false)
+    assert(securityManager.checkModifyPermissions("user1") === false)
+
+    // set groups only
+    securityManager.setAdminAclsGroups("group1,group2")
+    assert(securityManager.checkUIViewPermissions("user1") === false)
+    assert(securityManager.checkModifyPermissions("user1") === false)
+  }
+
   test("ssl on setup") {
     val conf = SSLSampleConfigs.sparkSSLConfig()
     val expectedAlgorithms = Set(