HDFS plugin: allow webhdfs scheme

plugins/repository-hdfs/build.gradle

@@ -137,6 +137,10 @@ for (int hadoopVersion = minTestedHadoopVersion; hadoopVersion <= maxTestedHadoo
             } else {
                 miniHDFSArgs.add("-Dhdfs.config.port=" + getNonSecureNamenodePortForVersion(hadoopVer))
             }
+            // This is required for webhdfs to work, otherwise miniHDFS will fail with module java.base does not "opens java.lang" to unnamed module
+            miniHDFSArgs.add('--add-opens')
+            miniHDFSArgs.add('java.base/java.lang=ALL-UNNAMED')
+            miniHDFSArgs.add('-Dhdfs.config.http.port=' + getHttpPortForVersion(hadoopVer))
             // If it's an HA fixture, set a nameservice to use in the JVM options
             if (name.startsWith('haHdfs') || name.startsWith('secureHaHdfs')) {
                 miniHDFSArgs.add("-Dha-nameservice=ha-hdfs")
@@ -189,7 +193,7 @@ for (int hadoopVersion = minTestedHadoopVersion; hadoopVersion <= maxTestedHadoo
         }
     }
 
-    for (String integTestTaskName : ['yamlRestTest' + hadoopVersion, 'yamlRestTestSecure' + hadoopVersion]) {
+    for (String integTestTaskName : ['yamlRestTest' + hadoopVersion, 'yamlRestTestSecure' + hadoopVersion, 'yamlRestTestWeb' + hadoopVersion]) {
         tasks.register(integTestTaskName, RestIntegTestTask) {
             description = "Runs rest tests against an elasticsearch cluster with HDFS" + hadoopVer
 
@@ -207,6 +211,7 @@ for (int hadoopVersion = minTestedHadoopVersion; hadoopVersion <= maxTestedHadoo
         Map<String, Object> expansions = [
                 'hdfs_port'       : getNonSecureNamenodePortForVersion(hadoopVer),
                 'secure_hdfs_port': getSecureNamenodePortForVersion(hadoopVer),
+                'hdfs_http_port'  : getHttpPortForVersion(hadoopVer),
         ]
         inputs.properties(expansions)
         filter("tokens": expansions.collectEntries { k, v -> [k, v.toString()]}, ReplaceTokens.class)
@@ -217,6 +222,9 @@ for (int hadoopVersion = minTestedHadoopVersion; hadoopVersion <= maxTestedHadoo
         it.into("secure_hdfs_repository_" + hadoopVer) {
             from "src/yamlRestTest/resources/rest-api-spec/test/secure_hdfs_repository"
         }
+        it.into("webhdfs_repository_" + hadoopVer) {
+          from "src/yamlRestTest/resources/rest-api-spec/test/webhdfs_repository"
+        }
     }
     tasks.named("processYamlRestTestResources").configure {
         dependsOn(processHadoopTestResources)
@@ -234,6 +242,11 @@ for (int hadoopVersion = minTestedHadoopVersion; hadoopVersion <= maxTestedHadoo
             // The normal test runner only runs the standard hdfs rest tests
             systemProperty 'tests.rest.suite', 'hdfs_repository_' + hadoopVer
         }
+        tasks.named("yamlRestTestWeb" + hadoopVer).configure {
+          dependsOn "hdfs" + hadoopVer + "Fixture"
+          // The normal test runner only runs the standard hdfs rest tests
+          systemProperty 'tests.rest.suite', 'hdfs_repository_' + hadoopVer
+        }
         tasks.named("javaRestTest" + hadoopVer).configure {
             dependsOn "haHdfs" + hadoopVer + "Fixture"
         }
@@ -243,19 +256,27 @@ for (int hadoopVersion = minTestedHadoopVersion; hadoopVersion <= maxTestedHadoo
             systemProperty 'tests.rest.suite', 'hdfs_repository_' + hadoopVer + '/10_basic'
         }
         // HA fixture is unsupported. Don't run them.
+        tasks.named("yamlRestTestSecure" + hadoopVer).configure {
+          enabled = false
+        }
         tasks.named("javaRestTestSecure" + hadoopVer).configure {
             enabled = false
         }
     }
 
     tasks.named("check").configure {
-        dependsOn("yamlRestTest" + hadoopVer, "yamlRestTestSecure" + hadoopVer, "javaRestTestSecure" + hadoopVer)
+        dependsOn("yamlRestTest" + hadoopVer, "yamlRestTestSecure" + hadoopVer, "javaRestTestSecure" + hadoopVer, "yamlRestTestWeb" + hadoopVer)
     }
 
     // Run just the secure hdfs rest test suite.
     tasks.named("yamlRestTestSecure" + hadoopVer).configure {
         systemProperty 'tests.rest.suite', 'secure_hdfs_repository_' + hadoopVer
     }
+
+  // Run just the secure webhdfs rest test suite.
+  tasks.named("yamlRestTestWeb" + hadoopVer).configure {
+    systemProperty 'tests.rest.suite', 'webhdfs_repository_' + hadoopVer
+  }
 }
 
 
@@ -267,6 +288,10 @@ def getNonSecureNamenodePortForVersion(hadoopVersion) {
     return 10003 - (2 * hadoopVersion)
 }
 
+def getHttpPortForVersion(hadoopVersion) {
+  return 10004 - (2 * hadoopVersion)
+}
+
 Set disabledIntegTestTaskNames = []
 
 tasks.withType(RestIntegTestTask).configureEach { testTask ->

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

@@ -35,7 +35,11 @@ final class HdfsBlobStore implements BlobStore {
         this.fileContext = fileContext;
         // Only restrict permissions if not running with HA
         boolean restrictPermissions = (haEnabled == false);
-        this.securityContext = new HdfsSecurityContext(fileContext.getUgi(), restrictPermissions);
+        this.securityContext = new HdfsSecurityContext(
+            fileContext.getUgi(),
+            restrictPermissions,
+            fileContext.getDefaultFileSystem().getUri().getScheme()
+        );
         this.bufferSize = bufferSize;
         this.root = execute(fileContext1 -> fileContext1.makeQualified(new Path(path)));
         this.readOnly = readOnly;

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

@@ -39,6 +39,7 @@
 import java.net.UnknownHostException;
 import java.security.AccessController;
 import java.security.PrivilegedAction;
+import java.util.List;
 import java.util.Locale;
 
 public final class HdfsRepository extends BlobStoreRepository {
@@ -47,6 +48,8 @@ public final class HdfsRepository extends BlobStoreRepository {
 
     private static final String CONF_SECURITY_PRINCIPAL = "security.principal";
 
+    private static final List<String> allowedSchemes = List.of("hdfs", "webhdfs", "swebhdfs");
+
     private final Environment environment;
     private final ByteSizeValue chunkSize;
     private final URI uri;
@@ -70,17 +73,18 @@ public HdfsRepository(
             throw new IllegalArgumentException("No 'uri' defined for hdfs snapshot/restore");
         }
         uri = URI.create(uriSetting);
-        if ("hdfs".equalsIgnoreCase(uri.getScheme()) == false) {
+        if (uri.getScheme() == null || allowedSchemes.contains(uri.getScheme()) == false) {
             throw new IllegalArgumentException(
                 String.format(
                     Locale.ROOT,
-                    "Invalid scheme [%s] specified in uri [%s]; only 'hdfs' uri allowed for hdfs snapshot/restore",
+                    "Invalid scheme [%s] specified in uri [%s]; only one of %s uris allowed for hdfs snapshot/restore",
                     uri.getScheme(),
-                    uriSetting
+                    uriSetting,
+                    String.join(", ", allowedSchemes)
                 )
             );
         }
-        if (Strings.hasLength(uri.getPath()) && uri.getPath().equals("/") == false) {
+        if (uri.getScheme().equals("hdfs") && Strings.hasLength(uri.getPath()) && uri.getPath().equals("/") == false) {
             throw new IllegalArgumentException(
                 String.format(
                     Locale.ROOT,
@@ -111,6 +115,8 @@ private HdfsBlobStore createBlobstore(URI blobstoreUri, String path, Settings re
 
         // Disable FS cache
         hadoopConfiguration.setBoolean("fs.hdfs.impl.disable.cache", true);
+        hadoopConfiguration.setBoolean("fs.webhdfs.impl.disable.cache", true);
+        hadoopConfiguration.setBoolean("fs.swebhdfs.impl.disable.cache", true);
 
         // Create a hadoop user
         UserGroupInformation ugi = login(hadoopConfiguration, repositorySettings);

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

@@ -9,6 +9,7 @@
 
 import org.apache.hadoop.security.UserGroupInformation;
 import org.elasticsearch.SpecialPermission;
+import org.elasticsearch.common.util.ArrayUtils;
 import org.elasticsearch.env.Environment;
 
 import java.io.IOException;
@@ -22,6 +23,7 @@
 import java.security.PrivilegedActionException;
 import java.security.PrivilegedExceptionAction;
 import java.util.Arrays;
+import java.util.Objects;
 
 import javax.security.auth.AuthPermission;
 import javax.security.auth.PrivateCredentialPermission;
@@ -37,6 +39,8 @@ class HdfsSecurityContext {
 
     private static final Permission[] SIMPLE_AUTH_PERMISSIONS;
     private static final Permission[] KERBEROS_AUTH_PERMISSIONS;
+    private static final Permission[] WEBHDFS_PERMISSIONS;
+    private static final Permission[] SWEBHDFS_PERMISSIONS;
     static {
         // We can do FS ops with only a few elevated permissions:
         SIMPLE_AUTH_PERMISSIONS = new Permission[] {
@@ -72,6 +76,16 @@ class HdfsSecurityContext {
             // 7) allow code to initiate kerberos connections as the logged in user
             // Still far and away fewer permissions than the original full plugin policy
         };
+        WEBHDFS_PERMISSIONS = new Permission[] {
+            // 1) allow hadoop to act as the logged in Subject
+            new AuthPermission("doAs")
+        };
+        SWEBHDFS_PERMISSIONS = new Permission[] {
+            // 1) allow hadoop to act as the logged in Subject
+            new AuthPermission("doAs"),
+            // 2) allow hadoop to call setSSLSocketFactory on HttpsURLConnection
+            new RuntimePermission("setFactory"),
+        };
     }
 
     /**
@@ -94,13 +108,13 @@ static Path locateKeytabFile(Environment environment) {
     private final boolean restrictPermissions;
     private final Permission[] restrictedExecutionPermissions;
 
-    HdfsSecurityContext(UserGroupInformation ugi, boolean restrictPermissions) {
+    HdfsSecurityContext(UserGroupInformation ugi, boolean restrictPermissions, String scheme) {
         this.ugi = ugi;
         this.restrictPermissions = restrictPermissions;
-        this.restrictedExecutionPermissions = renderPermissions(ugi);
+        this.restrictedExecutionPermissions = renderPermissions(ugi, scheme);
     }
 
-    private Permission[] renderPermissions(UserGroupInformation userGroupInformation) {
+    private Permission[] renderPermissions(UserGroupInformation userGroupInformation, String scheme) {
         Permission[] permissions;
         if (userGroupInformation.isFromKeytab()) {
             // KERBEROS
@@ -117,6 +131,12 @@ private Permission[] renderPermissions(UserGroupInformation userGroupInformation
             // SIMPLE
             permissions = Arrays.copyOf(SIMPLE_AUTH_PERMISSIONS, SIMPLE_AUTH_PERMISSIONS.length);
         }
+        if (Objects.equals(scheme, "webhdfs")) {
+            permissions = ArrayUtils.concat(permissions, WEBHDFS_PERMISSIONS, Permission.class);
+        } else if (Objects.equals(scheme, "swebhdfs")) {
+            permissions = ArrayUtils.concat(permissions, SWEBHDFS_PERMISSIONS, Permission.class);
+        }
+
         return permissions;
     }
 

plugins/repository-hdfs/src/main/plugin-metadata/plugin-security.policy

@@ -68,4 +68,8 @@ grant {
   // client binds to the address returned from the host name of any principal set up as a service principal
   // org.apache.hadoop.ipc.Client.Connection.setupConnection
   permission java.net.SocketPermission "localhost:0", "listen,resolve";
+
+  // org.apache.hadoop.hdfs.web.SSLConnectionConfigurator
+  // This is used by swebhdfs connections
+  permission java.lang.RuntimePermission "setFactory";
 };

plugins/repository-hdfs/src/test/java/org/elasticsearch/repositories/hdfs/HdfsTests.java

@@ -196,6 +196,36 @@ public void testPathSpecifiedInHdfs() {
         }
     }
 
+    public void testWebhdfsIsAllowedScheme() {
+        client().admin()
+            .cluster()
+            .preparePutRepository("test-repo")
+            .setType("hdfs")
+            .setSettings(
+                Settings.builder()
+                    .put("uri", "webhdfs:///")
+                    .put("conf.fs.AbstractFileSystem.webhdfs.impl", TestingFs.class.getName())
+                    .put("path", "foo")
+                    .put("chunk_size", randomIntBetween(100, 1000) + "k")
+                    .put("compress", randomBoolean())
+            )
+            .get();
+    }
+
+    public void testSwebhdfsIsAllowedScheme() {
+        client().admin()
+            .cluster()
+            .preparePutRepository("test-repo")
+            .setType("hdfs")
+            .setSettings(
+                Settings.builder()
+                    .put("uri", "swebhdfs:///")
+                    .put("conf.fs.AbstractFileSystem.swebhdfs.impl", TestingFs.class.getName())
+                    .put("path", "foo")
+            )
+            .get();
+    }
+
     public void testMissingPath() {
         try {
             client().admin()

plugins/repository-hdfs/src/yamlRestTest/resources/rest-api-spec/test/hdfs_repository/10_basic.yml

@@ -19,7 +19,7 @@
 ---
 #
 # Check that we can't use file:// repositories or anything like that
-# We only test this plugin against hdfs://
+# We only test this plugin against hdfs:// and webhdfs://
 #
 "HDFS only":
     - do:

plugins/repository-hdfs/src/yamlRestTest/resources/rest-api-spec/test/secure_hdfs_repository/10_basic.yml

@@ -19,7 +19,7 @@
 ---
 #
 # Check that we can't use file:// repositories or anything like that
-# We only test this plugin against hdfs://
+# We only test this plugin against hdfs:// and webhdfs://
 #
 "HDFS only":
     - do:

plugins/repository-hdfs/src/yamlRestTest/resources/rest-api-spec/test/webhdfs_repository/10_basic.yml

@@ -0,0 +1,33 @@
+# Integration tests for HDFS Repository plugin
+#
+# Check plugin is installed
+#
+"Plugin loaded":
+    - skip:
+        reason: "contains is a newly added assertion"
+        features: contains
+    - do:
+        cluster.state: {}
+
+    # Get master node id
+    - set: { master_node: master }
+
+    - do:
+        nodes.info: {}
+
+    - contains:  { nodes.$master.plugins: { name: repository-hdfs } }
+---
+#
+# Check that we can't use file:// repositories or anything like that
+# We only test this plugin against hdfs:// and webhdfs://
+#
+"HDFS only":
+    - do:
+        catch: /Invalid scheme/
+        snapshot.create_repository:
+          repository: misconfigured_repository
+          body:
+            type: hdfs
+            settings:
+              uri: "file://bogus"
+              path: "foo/bar"

plugins/repository-hdfs/src/yamlRestTest/resources/rest-api-spec/test/webhdfs_repository/20_repository_create.yml

@@ -0,0 +1,27 @@
+# Integration tests for HDFS Repository plugin
+#
+# Tests creating a repository
+#
+"HDFS Repository Creation":
+    # Create repository
+    - do:
+        snapshot.create_repository:
+          repository: test_repository_create
+          body:
+            type: hdfs
+            settings:
+              uri: "webhdfs://localhost:@hdfs_http_port@"
+              path: "test/repository_create"
+
+    # Get repository
+    - do:
+        snapshot.get_repository:
+          repository: test_repository_create
+
+    - is_true: test_repository_create
+    - match: {test_repository_create.settings.path : "test/repository_create"}
+
+    # Remove our repository
+    - do:
+       snapshot.delete_repository:
+         repository: test_repository_create