Add permission checks before reading from HDFS stream

plugins/repository-hdfs/src/main/java/org/elasticsearch/repositories/hdfs/HdfsBlobContainer.java

@@ -23,6 +23,7 @@
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.Options.CreateOpts;
 import org.apache.hadoop.fs.Path;
+import org.elasticsearch.SpecialPermission;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.blobstore.BlobMetaData;
 import org.elasticsearch.common.blobstore.BlobPath;
@@ -45,12 +46,14 @@
 
 final class HdfsBlobContainer extends AbstractBlobContainer {
     private final HdfsBlobStore store;
+    private final HdfsSecurityContext securityContext;
     private final Path path;
     private final int bufferSize;
 
-    HdfsBlobContainer(BlobPath blobPath, HdfsBlobStore store, Path path, int bufferSize) {
+    HdfsBlobContainer(BlobPath blobPath, HdfsBlobStore store, Path path, int bufferSize, HdfsSecurityContext hdfsSecurityContext) {
         super(blobPath);
         this.store = store;
+        this.securityContext = hdfsSecurityContext;
         this.path = path;
         this.bufferSize = bufferSize;
     }
@@ -90,7 +93,9 @@ public InputStream readBlob(String blobName) throws IOException {
         // FSDataInputStream can open connections on read() or skip() so we wrap in
         // HDFSPrivilegedInputSteam which will ensure that underlying methods will
         // be called with the proper privileges.
-        return store.execute(fileContext -> new HDFSPrivilegedInputSteam(fileContext.open(new Path(path, blobName), bufferSize)));
+        return store.execute(fileContext ->
+            new HDFSPrivilegedInputSteam(fileContext.open(new Path(path, blobName), bufferSize), securityContext)
+        );
     }
 
     @Override
@@ -144,8 +149,11 @@ public Map<String, BlobMetaData> listBlobs() throws IOException {
      */
     private static class HDFSPrivilegedInputSteam extends FilterInputStream {
 
-        HDFSPrivilegedInputSteam(InputStream in) {
+        private final HdfsSecurityContext securityContext;
+
+        HDFSPrivilegedInputSteam(InputStream in, HdfsSecurityContext hdfsSecurityContext) {
             super(in);
+            this.securityContext = hdfsSecurityContext;
         }
 
         public int read() throws IOException {
@@ -175,9 +183,10 @@ public synchronized void reset() throws IOException {
             });
         }
 
-        private static  <T> T doPrivilegedOrThrow(PrivilegedExceptionAction<T> action) throws IOException {
+        private <T> T doPrivilegedOrThrow(PrivilegedExceptionAction<T> action) throws IOException {
+            SpecialPermission.check();
             try {
-                return AccessController.doPrivileged(action);
+                return AccessController.doPrivileged(action, null, securityContext.getRestrictedExecutionPermissions());
             } catch (PrivilegedActionException e) {
                 throw (IOException) e.getCause();
             }

plugins/repository-hdfs/src/main/java/org/elasticsearch/repositories/hdfs/HdfsBlobStore.java

@@ -75,7 +75,7 @@ public String toString() {
 
     @Override
     public BlobContainer blobContainer(BlobPath path) {
-        return new HdfsBlobContainer(path, this, buildHdfsPath(path), bufferSize);
+        return new HdfsBlobContainer(path, this, buildHdfsPath(path), bufferSize, securityContext);
     }
 
     private Path buildHdfsPath(BlobPath blobPath) {

plugins/repository-hdfs/src/main/java/org/elasticsearch/repositories/hdfs/HdfsSecurityContext.java

@@ -56,7 +56,9 @@ class HdfsSecurityContext {
             // 1) hadoop dynamic proxy is messy with access rules
             new ReflectPermission("suppressAccessChecks"),
             // 2) allow hadoop to add credentials to our Subject
-            new AuthPermission("modifyPrivateCredentials")
+            new AuthPermission("modifyPrivateCredentials"),
+            // 3) RPC Engine requires this for re-establishing pooled connections over the lifetime of the client
+            new PrivateCredentialPermission("org.apache.hadoop.security.Credentials * \"*\"", "read")
         };
 
         // If Security is enabled, we need all the following elevated permissions: