Merge remote-tracking branch 'upstream/master'

Author: gatorsmile

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,24 +198,37 @@ 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())
 
   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
@@ -302,17 +320,34 @@ private[spark] class SecurityManager(sparkConf: SparkConf)
     setViewAcls(Set[String](defaultUser), allowedUsers)
   }
 
+  /**
+   * 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.
@@ -322,17 +357,34 @@ private[spark] class SecurityManager(sparkConf: SparkConf)
     logInfo("Changing modify acls to: " + modifyAcls.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 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

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))
           }
         }

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")
     }
   }
 }

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]
+
+}

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
+  }
+}

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].