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

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

@@ -46,6 +46,7 @@ import java.security.Policy;
 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 @@ final class Security {
         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,7 +407,8 @@ final class Security {
     }
 
     /**
-     * Add access to path (and all files underneath it)
+     * 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

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

@@ -35,6 +35,9 @@ import java.security.PermissionCollection;
 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 class EvilSecurityTests extends ESTestCase {
         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();