ZOOKEEPER-2804: Node creation fails with NPE if ACLs are null

1) Handled Null case in server. Client will get InvalidACLException
2) Handled null check in create and setACL APIs in client side as mentioned in their javadoc
throws KeeperException.InvalidACLException if the ACL is invalid, null, or empty
3) Not handling any validation for async API of create and setACL in this JIRA because these API doesn't throw KeeperException explicitly. So can not throw InvalidACLExceptin from Client.  If we throw IllegalArgumentException then it will not be consistent with other sync APIs.  So Let server throw InvalidACLException for async API.

Please review and provide suggestion.

Author: bhupendra jain <bhupendra.jain@huawei.com>

Reviewers: Mohammad Arshad <arshad@apache.org>, Michael Han <hanm@apache.org>

Closes #279 from jainbhupendra24/ZOOKEEPER-2804-new

src/java/main/org/apache/zookeeper/ZooKeeper.java

@@ -1429,6 +1429,7 @@ public class ZooKeeper implements AutoCloseable {
         final String clientPath = path;
         PathUtils.validatePath(clientPath, createMode.isSequential());
         EphemeralType.validateTTL(createMode, -1);
+        validateACL(acl);
 
         final String serverPath = prependChroot(clientPath);
 
@@ -1439,9 +1440,6 @@ public class ZooKeeper implements AutoCloseable {
         request.setData(data);
         request.setFlags(createMode.toFlag());
         request.setPath(serverPath);
-        if (acl != null && acl.size() == 0) {
-            throw new KeeperException.InvalidACLException();
-        }
         request.setAcl(acl);
         ReplyHeader r = cnxn.submitRequest(h, request, response, null);
         if (r.getErr() != 0) {
@@ -1532,15 +1530,13 @@ public class ZooKeeper implements AutoCloseable {
         final String clientPath = path;
         PathUtils.validatePath(clientPath, createMode.isSequential());
         EphemeralType.validateTTL(createMode, ttl);
+        validateACL(acl);
 
         final String serverPath = prependChroot(clientPath);
 
         RequestHeader h = new RequestHeader();
         setCreateHeader(createMode, h);
         Create2Response response = new Create2Response();
-        if (acl != null && acl.size() == 0) {
-            throw new KeeperException.InvalidACLException();
-        }
         Record record = makeCreateRecord(createMode, serverPath, data, acl, ttl);
         ReplyHeader r = cnxn.submitRequest(h, record, response, null);
         if (r.getErr() != 0) {
@@ -2373,6 +2369,7 @@ public class ZooKeeper implements AutoCloseable {
     {
         final String clientPath = path;
         PathUtils.validatePath(clientPath);
+        validateACL(acl);
 
         final String serverPath = prependChroot(clientPath);
 
@@ -2380,9 +2377,6 @@ public class ZooKeeper implements AutoCloseable {
         h.setType(ZooDefs.OpCode.setACL);
         SetACLRequest request = new SetACLRequest();
         request.setPath(serverPath);
-        if (acl != null && acl.size() == 0) {
-            throw new KeeperException.InvalidACLException(clientPath);
-        }
         request.setAcl(acl);
         request.setVersion(aclVersion);
         SetACLResponse response = new SetACLResponse();
@@ -2945,4 +2939,18 @@ public class ZooKeeper implements AutoCloseable {
             throw ioe;
         }
     }
+
+    /**
+     * Validates the provided ACL list for null, empty or null value in it.
+     * 
+     * @param acl
+     *            ACL list
+     * @throws InvalidACLException
+     *             if ACL list is not valid
+     */
+    private void validateACL(List<ACL> acl) throws KeeperException.InvalidACLException {
+        if (acl == null || acl.isEmpty() || acl.contains(null)) {
+            throw new KeeperException.InvalidACLException();
+        }
+    }
 }

src/java/main/org/apache/zookeeper/server/PrepRequestProcessor.java

@@ -912,9 +912,11 @@ public class PrepRequestProcessor extends ZooKeeperCriticalThread implements
     private List<ACL> removeDuplicates(List<ACL> acl) {
 
         LinkedList<ACL> retval = new LinkedList<ACL>();
-        for (ACL a : acl) {
-            if (!retval.contains(a)) {
-                retval.add(a);
+        if (acl != null) {
+            for (ACL a : acl) {
+                if (!retval.contains(a)) {
+                    retval.add(a);
+                }
             }
         }
         return retval;

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

@@ -22,11 +22,13 @@ import static org.apache.zookeeper.test.ClientBase.CONNECTION_TIMEOUT;
 
 import java.io.File;
 import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.CountDownLatch;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException.InvalidACLException;
 import org.apache.zookeeper.PortAssignment;
 import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.Watcher;
@@ -189,4 +191,92 @@ public class ACLTest extends ZKTestCase implements Watcher {
             }
         }
     }
+    
+    @Test
+    public void testNullACL() 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);
+        ZooKeeper zk = ClientBase.createZKClient(HOSTPORT);
+        try {
+            // case 1 : null ACL with create
+            try {
+                zk.create("/foo", "foo".getBytes(), null, CreateMode.PERSISTENT);
+                Assert.fail("Expected InvalidACLException for null ACL parameter");
+            } catch (InvalidACLException e) {
+                // Expected. Do nothing
+            }
+
+            // case 2 : null ACL with other create API
+            try {
+                zk.create("/foo", "foo".getBytes(), null, CreateMode.PERSISTENT, null);
+                Assert.fail("Expected InvalidACLException for null ACL parameter");
+            } catch (InvalidACLException e) {
+                // Expected. Do nothing
+            }
+            
+            // case 3 : null ACL with setACL
+            try {
+                zk.setACL("/foo", null, 0);
+                Assert.fail("Expected InvalidACLException for null ACL parameter");
+            } catch (InvalidACLException e) {
+                // Expected. Do nothing
+            }
+        } finally {
+            zk.close();
+            f.shutdown();
+            zks.shutdown();
+            Assert.assertTrue("waiting for server down",
+                    ClientBase.waitForServerDown(HOSTPORT, ClientBase.CONNECTION_TIMEOUT));
+        }
+    }
+
+    @Test
+    public void testNullValueACL() 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);
+        ZooKeeper zk = ClientBase.createZKClient(HOSTPORT);
+        try {
+
+            List<ACL> acls = new ArrayList<ACL>();
+            acls.add(null);
+
+            // case 1 : null value in ACL list with create
+            try {
+                zk.create("/foo", "foo".getBytes(), acls, CreateMode.PERSISTENT);
+                Assert.fail("Expected InvalidACLException for null value in ACL List");
+            } catch (InvalidACLException e) {
+                // Expected. Do nothing
+            }
+
+            // case 2 : null value in ACL list with other create API
+            try {
+                zk.create("/foo", "foo".getBytes(), acls, CreateMode.PERSISTENT, null);
+                Assert.fail("Expected InvalidACLException for null value in ACL List");
+            } catch (InvalidACLException e) {
+                // Expected. Do nothing
+            }
+
+            // case 3 : null value in ACL list with setACL
+            try {
+                zk.setACL("/foo", acls, -1);
+                Assert.fail("Expected InvalidACLException for null value in ACL List");
+            } catch (InvalidACLException e) {
+                // Expected. Do nothing
+            }
+        } finally {
+            zk.close();
+            f.shutdown();
+            zks.shutdown();
+            Assert.assertTrue("waiting for server down",
+                    ClientBase.waitForServerDown(HOSTPORT, ClientBase.CONNECTION_TIMEOUT));
+        }
+    }
 }