CURATOR-715. bottom up existence check for mkdirs

curator-client/src/main/java/org/apache/curator/utils/ZKPaths.java

@@ -289,7 +289,20 @@ public static void mkdirs(
             throws InterruptedException, KeeperException {
         PathUtils.validatePath(path);
 
-        int pos = 1; // skip first slash, root is guaranteed to exist
+        int pos = path.length();
+        String subPath;
+
+        // This first loop locates the first parent that doesn't exist from leaf nodes towards root
+        // this way, it is not required to have read access on all parents.
+        // This is relevant after https://issues.apache.org/jira/browse/ZOOKEEPER-2590.
+
+        do {
+            pos = path.lastIndexOf(PATH_SEPARATOR_CHAR, pos - 1);
+            subPath = path.substring(0, pos);
+        } while (pos > 0 && zookeeper.exists(subPath, false) == null);
+
+        // Start creating the subtree after the longest path that exists, assuming root always exists.
+
         do {
             pos = path.indexOf(PATH_SEPARATOR_CHAR, pos + 1);
 
@@ -301,23 +314,23 @@ public static void mkdirs(
                 }
             }
 
-            String subPath = path.substring(0, pos);
-            if (zookeeper.exists(subPath, false) == null) {
-                try {
-                    List<ACL> acl = null;
-                    if (aclProvider != null) {
-                        acl = aclProvider.getAclForPath(subPath);
-                        if (acl == null) {
-                            acl = aclProvider.getDefaultAcl();
-                        }
-                    }
+            subPath = path.substring(0, pos);
+
+            // All the paths from the initial `pos` do not exist.
+            try {
+                List<ACL> acl = null;
+                if (aclProvider != null) {
+                    acl = aclProvider.getAclForPath(subPath);
                     if (acl == null) {
-                        acl = ZooDefs.Ids.OPEN_ACL_UNSAFE;
+                        acl = aclProvider.getDefaultAcl();
                     }
-                    zookeeper.create(subPath, new byte[0], acl, getCreateMode(asContainers));
-                } catch (KeeperException.NodeExistsException ignore) {
-                    // ignore... someone else has created it since we checked
                 }
+                if (acl == null) {
+                    acl = ZooDefs.Ids.OPEN_ACL_UNSAFE;
+                }
+                zookeeper.create(subPath, new byte[0], acl, getCreateMode(asContainers));
+            } catch (KeeperException.NodeExistsException ignore) {
+                // ignore... someone else has created it since we checked
             }
 
         } while (pos < path.length());

curator-framework/src/test/java/org/apache/curator/framework/imps/TestCreate.java

@@ -20,9 +20,7 @@
 package org.apache.curator.framework.imps;
 
 import static org.apache.zookeeper.ZooDefs.Ids.ANYONE_ID_UNSAFE;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.*;
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
@@ -466,4 +464,35 @@ public void testProtectedUtils() throws Exception {
         path = ZKPaths.makePath("hola", name);
         assertEquals(ProtectedUtils.normalizePath(path), "/hola/yo");
     }
+
+    /**
+     * Tests that parents existence checks don't need READ access to the whole path (from / to the new node)
+     * but just to the first ancestor parent. (See https://issues.apache.org/jira/browse/ZOOKEEPER-2590)
+     */
+    @Test
+    public void testForbiddenAncestors() throws Exception {
+        CuratorFramework client = createClient(testACLProvider);
+        try {
+            client.start();
+
+            client.create().creatingParentsIfNeeded().forPath("/bat/bi/hiru");
+            client.setACL()
+                    .withACL(Collections.singletonList(new ACL(0, ANYONE_ID_UNSAFE)))
+                    .forPath("/bat");
+
+            // In creation attempts where the parent ("/bat") has ACL that restricts read, creation request fails.
+            try {
+                client.create().creatingParentsIfNeeded().forPath("/bat/bost");
+                fail("Expected NoAuthException when not authorized to read new node grandparent");
+            } catch (KeeperException.NoAuthException noAuthException) {
+            }
+
+            // But creating a node in the same subtree where its grandparent has read access is allowed and
+            // Curator will not check for higher nodes' existence.
+            client.create().creatingParentsIfNeeded().forPath("/bat/bi/hiru/bost");
+            assertNotNull(client.checkExists().forPath("/bat/bi/hiru/bost"));
+        } finally {
+            CloseableUtils.closeQuietly(client);
+        }
+    }
 }

pom.xml

@@ -72,7 +72,7 @@
         <redirectTestOutputToFile>true</redirectTestOutputToFile>
 
         <!-- versions -->
-        <zookeeper-version>3.9.1</zookeeper-version>
+        <zookeeper-version>3.9.2</zookeeper-version>
         <maven-bundle-plugin-version>5.1.4</maven-bundle-plugin-version>
         <maven-compiler-plugin-version>3.10.0</maven-compiler-plugin-version>
         <maven-dependency-plugin-version>3.2.0</maven-dependency-plugin-version>