ZOOKEEPER-2590: exists() should check read ACL permission

ZOOKEEPER-2590:exists() should check read ACL permission
ZOOKEEPER-2590. Skip ACL check if znode is missing
Reviewers: eolivelli
Author: anmolnar
Closes #2093 from anmolnar/ZOOKEEPER-2590

zookeeper-server/src/main/java/org/apache/zookeeper/server/FinalRequestProcessor.java

@@ -363,12 +363,21 @@ public class FinalRequestProcessor implements RequestProcessor {
             }
             case OpCode.exists: {
                 lastOp = "EXIS";
-                // TODO we need to figure out the security requirement for this!
                 ExistsRequest existsRequest = request.readRequestRecord(ExistsRequest::new);
                 path = existsRequest.getPath();
                 if (path.indexOf('\0') != -1) {
                     throw new KeeperException.BadArgumentsException();
                 }
+                DataNode n = zks.getZKDatabase().getNode(path);
+                if (n != null) {
+                    zks.checkACL(
+                        request.cnxn,
+                        zks.getZKDatabase().aclForNode(n),
+                        ZooDefs.Perms.READ,
+                        request.authInfo,
+                        path,
+                        null);
+                }
                 Stat stat = zks.getZKDatabase().statNode(path, existsRequest.getWatch() ? cnxn : null);
                 rsp = new ExistsResponse(stat);
                 requestPathMetricsCollector.registerRequest(request.type, path);

zookeeper-server/src/test/java/org/apache/zookeeper/test/ACLTest.java

@@ -19,7 +19,10 @@
 package org.apache.zookeeper.test;
 
 import static org.apache.zookeeper.test.ClientBase.CONNECTION_TIMEOUT;
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 import java.io.File;
@@ -27,16 +30,19 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.KeeperException.InvalidACLException;
 import org.apache.zookeeper.PortAssignment;
 import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.Watcher;
 import org.apache.zookeeper.Watcher.Event.KeeperState;
 import org.apache.zookeeper.ZKTestCase;
+import org.apache.zookeeper.ZooDefs;
 import org.apache.zookeeper.ZooDefs.Ids;
 import org.apache.zookeeper.ZooKeeper;
 import org.apache.zookeeper.data.ACL;
 import org.apache.zookeeper.data.Id;
+import org.apache.zookeeper.data.Stat;
 import org.apache.zookeeper.server.ServerCnxn;
 import org.apache.zookeeper.server.ServerCnxnFactory;
 import org.apache.zookeeper.server.SyncRequestProcessor;
@@ -302,4 +308,91 @@ public class ACLTest extends ZKTestCase implements Watcher {
         }
     }
 
+    @Test
+    public void testExistACLCheck() throws Exception {
+        File tmpDir = ClientBase.createTmpDir();
+        ClientBase.setupTestEnv();
+        ZooKeeperServer zks = new ZooKeeperServer(tmpDir, tmpDir, 3000);
+        final int PORT = Integer.parseInt(HOSTPORT.split(":")[1]);
+        ServerCnxnFactory f = ServerCnxnFactory.createFactory(PORT, -1);
+        f.startup(zks);
+        String path = "/testExistACLCheck";
+        String data = "/testExistACLCheck-data";
+        try {
+            LOG.info("starting up the zookeeper server .. waiting");
+            assertTrue(ClientBase.waitForServerUp(HOSTPORT, CONNECTION_TIMEOUT), "waiting for server being up");
+            ZooKeeper zk = ClientBase.createZKClient(HOSTPORT);
+            try {
+                Stat stat = zk.exists(path, false);
+                assertNull(stat);
+                zk.create(path, data.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+                stat = zk.exists(path, false);
+                assertNotNull(stat);
+                assertEquals(data.length(), stat.getDataLength());
+
+                zk.delete(path, -1);
+                ArrayList<ACL> acls = new ArrayList<>();
+                acls.add(new ACL(ZooDefs.Perms.WRITE, Ids.ANYONE_ID_UNSAFE));
+                zk.create(path, data.getBytes(), acls, CreateMode.PERSISTENT);
+                try {
+                    stat = zk.exists(path, false);
+                    fail("exists should throw NoAuthException when don't have read permission");
+                } catch (KeeperException.NoAuthException e) {
+                    //expected
+                }
+
+                zk.delete(path, -1);
+                acls = new ArrayList<>();
+                acls.add(new ACL(ZooDefs.Perms.READ, Ids.ANYONE_ID_UNSAFE));
+                zk.create(path, data.getBytes(), acls, CreateMode.PERSISTENT);
+                stat = zk.exists(path, false);
+                assertNotNull(stat);
+                assertEquals(data.length(), stat.getDataLength());
+            } finally {
+                zk.close();
+            }
+        } finally {
+            f.shutdown();
+            zks.shutdown();
+            assertTrue(ClientBase.waitForServerDown(HOSTPORT, ClientBase.CONNECTION_TIMEOUT), "waiting for server down");
+        }
+    }
+
+    @Test
+    public void testExistACLCheckAtRootPath() throws Exception {
+        File tmpDir = ClientBase.createTmpDir();
+        ClientBase.setupTestEnv();
+        ZooKeeperServer zks = new ZooKeeperServer(tmpDir, tmpDir, 3000);
+        final int PORT = Integer.parseInt(HOSTPORT.split(":")[1]);
+        ServerCnxnFactory f = ServerCnxnFactory.createFactory(PORT, -1);
+        f.startup(zks);
+        try {
+            LOG.info("starting up the zookeeper server .. waiting");
+            assertTrue(ClientBase.waitForServerUp(HOSTPORT, CONNECTION_TIMEOUT), "waiting for server being up");
+            ZooKeeper zk = ClientBase.createZKClient(HOSTPORT);
+            try {
+                String data = "/testExistACLCheckAtRootPath-data";
+                zk.create("/a", data.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+                ArrayList<ACL> acls = new ArrayList<>();
+                acls.add(new ACL(0, Ids.ANYONE_ID_UNSAFE));
+                zk.setACL("/", acls, -1);
+
+                Stat stat = zk.exists("/a", false);
+                assertNotNull(stat);
+                assertEquals(data.length(), stat.getDataLength());
+                try {
+                    stat = zk.exists("/", false);
+                    fail("exists should throw NoAuthException when removing root path's ACL permission");
+                } catch (KeeperException.NoAuthException e) {
+                    //expected
+                }
+            } finally {
+                zk.close();
+            }
+        } finally {
+            f.shutdown();
+            zks.shutdown();
+            assertTrue(ClientBase.waitForServerDown(HOSTPORT, ClientBase.CONNECTION_TIMEOUT), "waiting for server down");
+        }
+    }
 }