[SPARK-24948][SHS] Delegate check access permissions to the file system

Author: mgaido91

core/src/main/scala/org/apache/spark/deploy/SparkHadoopUtil.scala

@@ -26,6 +26,7 @@ import scala.collection.JavaConverters._
 import scala.collection.immutable.Map
 import scala.collection.mutable
 import scala.collection.mutable.HashMap
+import scala.util.Try
 import scala.util.control.NonFatal
 
 import com.google.common.primitives.Longs
@@ -383,6 +384,10 @@ class SparkHadoopUtil extends Logging {
       return true
     }
 
+    // We may still be able to access the file as ACL may be enabled or spark may be an admin user
+    if (Try(status.getPath.getFileSystem(conf).access(status.getPath, mode)).isSuccess) {
+      return true
+    }
     logDebug(s"Permission denied: user=${ugi.getShortUserName}, " +
       s"path=${status.getPath}:${status.getOwner}:${status.getGroup}" +
       s"${if (status.isDirectory) "d" else "-"}$perm")

core/src/test/scala/org/apache/spark/deploy/SparkHadoopUtilSuite.scala

@@ -21,9 +21,12 @@ import java.security.PrivilegedExceptionAction
 
 import scala.util.Random
 
-import org.apache.hadoop.fs.FileStatus
+import org.apache.hadoop.fs.{FileStatus, FileSystem, Path}
 import org.apache.hadoop.fs.permission.{FsAction, FsPermission}
-import org.apache.hadoop.security.UserGroupInformation
+import org.apache.hadoop.security.{AccessControlException, UserGroupInformation}
+import org.mockito
+import org.mockito.internal.stubbing.answers.DoesNothing
+import org.mockito.Mockito._
 import org.scalatest.Matchers
 
 import org.apache.spark.SparkFunSuite
@@ -72,6 +75,14 @@ class SparkHadoopUtilSuite extends SparkFunSuite with Matchers {
         sparkHadoopUtil.checkAccessPermission(status, READ) should be(false)
         sparkHadoopUtil.checkAccessPermission(status, WRITE) should be(false)
 
+        status = mockedStatus(otherUser, otherGroup, READ_WRITE, READ_WRITE, NONE, true)
+        sparkHadoopUtil.checkAccessPermission(status, READ) should be(true)
+        sparkHadoopUtil.checkAccessPermission(status, WRITE) should be(true)
+
+        status = mockedStatus(otherUser, otherGroup, READ_WRITE, READ_WRITE, NONE, false)
+        sparkHadoopUtil.checkAccessPermission(status, READ) should be(false)
+        sparkHadoopUtil.checkAccessPermission(status, WRITE) should be(false)
+
         null
       }
     })
@@ -94,4 +105,26 @@ class SparkHadoopUtilSuite extends SparkFunSuite with Matchers {
       group,
       null)
   }
+
+  private def mockedStatus(
+      owner: String,
+      group: String,
+      userAction: FsAction,
+      groupAction: FsAction,
+      otherAction: FsAction,
+      accessGranted: Boolean): FileStatus = {
+    val mockedFs = mock(classOf[FileSystem])
+    val stub = when(mockedFs.access(mockito.Matchers.any(), mockito.Matchers.any()))
+    if (accessGranted) {
+      stub.thenAnswer(new DoesNothing)
+    } else {
+      stub.thenThrow(new AccessControlException)
+    }
+    val mockedPath = mock(classOf[Path])
+    when(mockedPath.getFileSystem(mockito.Matchers.any())).thenReturn(mockedFs)
+    // This status has no access permission, here we modify only the Path it returns
+    val mockedStatus = spy(fileStatus(owner, group, userAction, groupAction, otherAction))
+    when(mockedStatus.getPath).thenReturn(mockedPath)
+    mockedStatus
+  }
 }