CURATOR-715: Check node existence bottom up in ZkPaths::mkdirs (#506)

ZOOKEEPER-2590 enforces read ACL permission for check(). 

When creating parents if needed, Apache Curator client checks the existence
of all nodes in the path from the root node to the created one. However,
this is not necessary, it is enough to check the existence of the nodes
between the new node and the first existing ancestor.

There are use cases where the first levels of a sub-tree are protected against
read through ACLs. The current implementation makes it impossible to use
`creatingParentsIfNeeded`.

This pr also bump ZooKeeper to 3.9.2 which ZOOKEEPER-2590 is shipped.

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;
@@ -597,4 +595,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>