Apache ZooKeeper: Insufficient Permission Check in AdminServer Snapshot/Restore Commands

@@ -227,6 +227,18 @@ private static List<Id> handleAuthentication(final HttpServletRequest request, f
         }
     }
 
+    /**
+     * Grant or deny authorization for a command by matching
+     * request-provided credentials with the ACLs present on a node.
+     *
+     * @param zkServer the ZooKeeper server object.
+     * @param ids the credentials extracted from the Authorization header.
+     * @param perm the set of permission bits required by the command.
+     * @param path the ZooKeeper node path whose ACLs should be used
+     * to satisfy the perm bits.
+     * @throws KeeperException.NoAuthException if one or more perm
+     * bits could not be satisfied.
+     */
     private static void handleAuthorization(final ZooKeeperServer zkServer,
                                             final List<Id> ids,
                                             final int perm,
@@ -237,7 +249,14 @@ private static void handleAuthorization(final ZooKeeperServer zkServer,
             throw new KeeperException.NoNodeException(path);
         }
         final List<ACL> acls = zkServer.getZKDatabase().aclForNode(dataNode);
-        zkServer.checkACL(null, acls, perm, ids, path, null);
+        // Check the individual bits of perm.
+        final int bitWidth = Integer.SIZE - Integer.numberOfLeadingZeros(perm);
+        for (int b = 0; b < bitWidth; b++) {
+            final int permBit = 1 << b;
+            if ((perm & permBit) != 0) {
+                zkServer.checkACL(null, acls, permBit, ids, path, null);
+            }
+        }
     }
 
     /**

@@ -29,9 +29,11 @@
 import java.net.HttpURLConnection;
 import java.net.URL;
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import javax.net.ssl.HttpsURLConnection;
 import javax.net.ssl.SSLContext;
@@ -185,6 +187,22 @@ public void testAuthCheck_noACL(final AuthSchema authSchema) throws Exception {
         assertEquals(HttpURLConnection.HTTP_OK, authTestConn.getResponseCode());
     }
 
+    @ParameterizedTest
+    @EnumSource(value = AuthSchema.class, names = {"DIGEST"})
+    public void testAuthCheck_noPerms(final AuthSchema authSchema) throws Exception {
+        // The extra ACL entry gives Perms.READ perms to the "invalid"
+        // DIGEST authInfo---but that should not permit access, as
+        // AuthTestCommand requires Perms.ADMIN.
+        setupRootACL(authSchema, ZooDefs.Ids.READ_ACL_UNSAFE);
+        try {
+            final HttpURLConnection authTestConn = sendAuthTestCommandRequest(authSchema, false);
+            assertEquals(HttpURLConnection.HTTP_FORBIDDEN, authTestConn.getResponseCode());
+        } finally {
+            addAuthInfo(zk, authSchema);
+            resetRootACL(zk);
+        }
+    }
+
     @Test
     public void testAuthCheck_invalidServerRequiredConfig() {
         assertThrows("An active server is required for auth check",
@@ -300,19 +318,29 @@ public void clearTLS() {
     }
 
     private void setupRootACL(final AuthSchema authSchema) throws Exception {
+        setupRootACL(authSchema, Collections.<ACL>emptyList());
+    }
+
+    private void setupRootACL(final AuthSchema authSchema, final List<ACL> extraEntries) throws Exception {
+        final List<ACL> aclEntries = new ArrayList<>();
+
         switch (authSchema) {
             case DIGEST:
-                setupRootACLForDigest(zk);
+                aclEntries.addAll(genACLForDigest());
                 break;
             case X509:
-                setupRootACLForX509(zk);
+                aclEntries.addAll(genACLForX509());
                 break;
             case IP:
-                setupRootACLForIP(zk);
+                aclEntries.addAll(genACLForIP());
                 break;
             default:
                 throw new IllegalArgumentException("Unknown auth schema");
         }
+
+        aclEntries.addAll(extraEntries);
+
+        zk.setACL(Commands.ROOT_PATH, aclEntries, -1);
     }
 
     private HttpURLConnection sendAuthTestCommandRequest(final AuthSchema authSchema, final boolean validAuthInfo) throws Exception  {
@@ -343,22 +371,22 @@ public static void resetRootACL(final ZooKeeper zk) throws Exception {
         zk.setACL(Commands.ROOT_PATH, OPEN_ACL_UNSAFE, -1);
     }
 
-    public static void setupRootACLForDigest(final ZooKeeper zk) throws Exception  {
+    public static List<ACL> genACLForDigest() throws Exception  {
         final String idPassword = String.format("%s:%s", ROOT_USER, ROOT_PASSWORD);
         final String digest = DigestAuthenticationProvider.generateDigest(idPassword);
 
         final ACL acl = new ACL(ZooDefs.Perms.ALL, new Id(DIGEST_SCHEMA, digest));
-        zk.setACL(Commands.ROOT_PATH, Collections.singletonList(acl), -1);
+        return Collections.singletonList(acl);
     }
 
-    private static void setupRootACLForX509(final ZooKeeper zk) throws Exception  {
+    private static List<ACL> genACLForX509() throws Exception  {
         final ACL acl = new ACL(ZooDefs.Perms.ALL, new Id(X509_SCHEMA, X509_SUBJECT_PRINCIPAL));
-        zk.setACL(Commands.ROOT_PATH, Collections.singletonList(acl), -1);
+        return Collections.singletonList(acl);
     }
 
-    private static void setupRootACLForIP(final ZooKeeper zk) throws Exception  {
+    private static List<ACL> genACLForIP() throws Exception  {
         final ACL acl = new ACL(ZooDefs.Perms.ALL, new Id(IP_SCHEMA, "127.0.0.1"));
-        zk.setACL(Commands.ROOT_PATH, Collections.singletonList(acl), -1);
+        return Collections.singletonList(acl);
     }
 
     public static void addAuthInfoForDigest(final ZooKeeper zk) {

@@ -21,7 +21,7 @@
 import static org.apache.zookeeper.server.ZooKeeperServer.ZOOKEEPER_SERIALIZE_LAST_PROCESSED_ZXID_ENABLED;
 import static org.apache.zookeeper.server.admin.CommandAuthTest.addAuthInfoForDigest;
 import static org.apache.zookeeper.server.admin.CommandAuthTest.resetRootACL;
-import static org.apache.zookeeper.server.admin.CommandAuthTest.setupRootACLForDigest;
+import static org.apache.zookeeper.server.admin.CommandAuthTest.genACLForDigest;
 import static org.apache.zookeeper.server.admin.Commands.ADMIN_RATE_LIMITER_INTERVAL;
 import static org.apache.zookeeper.server.admin.Commands.RestoreCommand.ADMIN_RESTORE_ENABLED;
 import static org.apache.zookeeper.server.admin.Commands.SnapshotCommand.ADMIN_SNAPSHOT_ENABLED;
@@ -119,7 +119,7 @@ public void setup() throws Exception {
         zk = ClientBase.createZKClient(hostPort);
 
         // setup root ACL
-        setupRootACLForDigest(zk);
+        zk.setACL(Commands.ROOT_PATH, genACLForDigest(), -1);
 
         // add auth
         addAuthInfoForDigest(zk);