Explicitly reject duplicate data paths

Duplicate data paths already fail to work because we would attempt to
take out a node lock on the directory a second time which will fail
after the first lock attempt succeeds. However, how this failure
manifests is not apparent at all and is quite difficult to
debug. Instead, we should explicitly reject duplicate data paths to make
the failure cause more obvious.

Relates #25178

Author: jasontedor

core/src/main/java/org/elasticsearch/bootstrap/Security.java

@@ -46,6 +46,7 @@
 import java.security.URIParameter;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.LinkedHashSet;
 import java.util.Map;
 import java.util.Set;
@@ -262,8 +263,22 @@ static void addFilePermissions(Permissions policy, Environment environment) {
         if (environment.sharedDataFile() != null) {
             addPath(policy, Environment.PATH_SHARED_DATA_SETTING.getKey(), environment.sharedDataFile(), "read,readlink,write,delete");
         }
+        final Set<Path> dataFilesPaths = new HashSet<>();
         for (Path path : environment.dataFiles()) {
             addPath(policy, Environment.PATH_DATA_SETTING.getKey(), path, "read,readlink,write,delete");
+            /*
+             * We have to do this after adding the path because a side effect of that is that the directory is created; the Path#toRealPath
+             * invocation will fail if the directory does not already exist. We use Path#toRealPath to follow symlinks and handle issues
+             * like unicode normalization or case-insensitivity on some filesystems (e.g., the case-insensitive variant of HFS+ on macOS).
+             */
+            try {
+                final Path realPath = path.toRealPath();
+                if (!dataFilesPaths.add(realPath)) {
+                    throw new IllegalStateException("path [" + realPath + "] is duplicated by [" + path + "]");
+                }
+            } catch (final IOException e) {
+                throw new IllegalStateException("unable to access [" + path + "]", e);
+            }
         }
         /*
          * If path.data and default.path.data are set, we need read access to the paths in default.path.data to check for the existence of
@@ -392,11 +407,12 @@ private static void addSocketPermissionForPortRange(final Permissions policy, fi
     }
 
     /**
-     * Add access to path (and all files underneath it)
-     * @param policy current policy to add permissions to
+     * Add access to path (and all files underneath it); this also creates the directory if it does not exist.
+     *
+     * @param policy            current policy to add permissions to
      * @param configurationName the configuration name associated with the path (for error messages only)
-     * @param path the path itself
-     * @param permissions set of file permissions to grant to the path
+     * @param path              the path itself
+     * @param permissions       set of file permissions to grant to the path
      */
     static void addPath(Permissions policy, String configurationName, Path path, String permissions) {
         // paths may not exist yet, this also checks accessibility

qa/evil-tests/src/test/java/org/elasticsearch/bootstrap/EvilSecurityTests.java

@@ -35,6 +35,9 @@
 import java.security.Permissions;
 import java.util.Set;
 
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.hasToString;
+
 @SuppressForbidden(reason = "modifies system properties and attempts to create symbolic links intentionally")
 public class EvilSecurityTests extends ESTestCase {
 
@@ -135,6 +138,30 @@ public void testEnvironmentPaths() throws Exception {
         assertExactPermissions(new FilePermission(environment.pidFile().toString(), "delete"), permissions);
     }
 
+    public void testDuplicateDataPaths() throws IOException {
+        final Path path = createTempDir();
+        final Path home = path.resolve("home");
+        final Path data = path.resolve("data");
+        final Path duplicate;
+        if (randomBoolean()) {
+            duplicate = data;
+        } else {
+            duplicate = createTempDir().toAbsolutePath().resolve("link");
+            Files.createSymbolicLink(duplicate, data);
+        }
+
+        final Settings settings =
+                Settings
+                        .builder()
+                        .put(Environment.PATH_HOME_SETTING.getKey(), home.toString())
+                        .putArray(Environment.PATH_DATA_SETTING.getKey(), data.toString(), duplicate.toString())
+                        .build();
+
+        final Environment environment = new Environment(settings);
+        final IllegalStateException e = expectThrows(IllegalStateException.class, () -> Security.createPermissions(environment));
+        assertThat(e, hasToString(containsString("path [" + duplicate.toRealPath() + "] is duplicated by [" + duplicate + "]")));
+    }
+
     public void testEnsureSymlink() throws IOException {
         Path p = createTempDir();