HBASE-2418 Support for ZooKeeper authentication

git-svn-id: https://svn.apache.org/repos/asf/hbase/trunk@1204227 13f79535-47bb-0310-9956-ffa450edef68

pom.xml

@@ -861,7 +861,7 @@
     <slf4j.version>1.5.8</slf4j.version><!-- newer version available -->
     <stax-api.version>1.0.1</stax-api.version>
     <thrift.version>0.7.0</thrift.version>
-    <zookeeper.version>3.3.3</zookeeper.version>
+    <zookeeper.version>3.4.0-SNAPSHOT</zookeeper.version>
     <hadoop-snappy.version>0.0.1-SNAPSHOT</hadoop-snappy.version>
 
     <package.prefix>/usr</package.prefix>	           
@@ -1404,6 +1404,9 @@
     <!-- profile for building against Hadoop 0.20+security-->
     <profile>
       <id>security</id>
+      <properties>
+        <hadoop.version>0.20.205.1-7070-SNAPSHOT</hadoop.version>
+      </properties>
       <build>
         <finalName>${project.artifactId}-${project.version}-security</finalName>
         <plugins>

src/main/java/org/apache/hadoop/hbase/zookeeper/MiniZooKeeperCluster.java

@@ -34,8 +34,10 @@ import java.util.Random;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileUtil;
-import org.apache.zookeeper.server.NIOServerCnxn;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.zookeeper.server.NIOServerCnxnFactory;
 import org.apache.zookeeper.server.ZooKeeperServer;
 import org.apache.zookeeper.server.persistence.FileTxnLog;
 
@@ -57,20 +59,26 @@ public class MiniZooKeeperCluster {
 
   private int clientPort;
 
-  private List<NIOServerCnxn.Factory> standaloneServerFactoryList;
+  private List<NIOServerCnxnFactory> standaloneServerFactoryList;
   private List<ZooKeeperServer> zooKeeperServers;
   private List<Integer> clientPortList;
 
   private int activeZKServerIndex;
   private int tickTime = 0;
 
-  /** Create mini ZooKeeper cluster. */
+  private Configuration configuration;
+
   public MiniZooKeeperCluster() {
+    this(new Configuration());
+  }
+
+  public MiniZooKeeperCluster(Configuration configuration) {
     this.started = false;
+    this.configuration = configuration;
     activeZKServerIndex = -1;
     zooKeeperServers = new ArrayList<ZooKeeperServer>();
     clientPortList = new ArrayList<Integer>();
-    standaloneServerFactoryList = new ArrayList<NIOServerCnxn.Factory>();
+    standaloneServerFactoryList = new ArrayList<NIOServerCnxnFactory>();
   }
 
   public void setDefaultClientPort(int clientPort) {
@@ -148,11 +156,14 @@ public class MiniZooKeeperCluster {
         tickTimeToUse = TICK_TIME;
       }
       ZooKeeperServer server = new ZooKeeperServer(dir, dir, tickTimeToUse);
-      NIOServerCnxn.Factory standaloneServerFactory;
+      NIOServerCnxnFactory standaloneServerFactory;
       while (true) {
         try {
-          standaloneServerFactory = new NIOServerCnxn.Factory(
+          standaloneServerFactory = new NIOServerCnxnFactory();
-              new InetSocketAddress(tentativePort));
+          standaloneServerFactory.configure(
+            new InetSocketAddress(tentativePort),
+            configuration.getInt(HConstants.ZOOKEEPER_MAX_CLIENT_CNXNS,
+              1000));
         } catch (BindException e) {
           LOG.debug("Failed binding ZK Server to client port: " +
               tentativePort);
@@ -204,7 +215,7 @@ public class MiniZooKeeperCluster {
     }
     // shut down all the zk servers
     for (int i = 0; i < standaloneServerFactoryList.size(); i++) {
-      NIOServerCnxn.Factory standaloneServerFactory =
+      NIOServerCnxnFactory standaloneServerFactory =
         standaloneServerFactoryList.get(i);
       int clientPort = clientPortList.get(i);
 
@@ -236,7 +247,7 @@ public class MiniZooKeeperCluster {
     }
 
     // Shutdown the current active one
-    NIOServerCnxn.Factory standaloneServerFactory =
+    NIOServerCnxnFactory standaloneServerFactory =
       standaloneServerFactoryList.get(activeZKServerIndex);
     int clientPort = clientPortList.get(activeZKServerIndex);
 
@@ -277,7 +288,7 @@ public class MiniZooKeeperCluster {
 
     int backupZKServerIndex = activeZKServerIndex+1;
     // Shutdown the current active one
-    NIOServerCnxn.Factory standaloneServerFactory =
+    NIOServerCnxnFactory standaloneServerFactory =
       standaloneServerFactoryList.get(backupZKServerIndex);
     int clientPort = clientPortList.get(backupZKServerIndex);
 

src/main/java/org/apache/hadoop/hbase/zookeeper/ZKUtil.java

@@ -40,9 +40,9 @@ import org.apache.zookeeper.AsyncCallback;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.Watcher;
-import org.apache.zookeeper.ZooKeeper;
 import org.apache.zookeeper.KeeperException.NoNodeException;
 import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.data.ACL;
 import org.apache.zookeeper.data.Stat;
 
 /**
@@ -696,6 +696,41 @@ public class ZKUtil {
     setData(zkw, znode, data, -1);
   }
 
+  public static boolean isSecureZooKeeper(Configuration conf) {
+    // TODO: We need a better check for security enabled ZooKeeper. Currently
+    // the secure ZooKeeper client is set up using a supplied JaaS
+    // configuration file. But if the system property for the JaaS
+    // configuration file is set, this may not be an exclusive indication
+    // that HBase should set ACLs on znodes. As an alternative, we could do
+    // this more like Hadoop and build a JaaS configuration programmatically
+    // based on a site conf setting. The scope of such a change will be
+    // addressed in HBASE-4791.
+    return (System.getProperty("java.security.auth.login.config") != null);
+  }
+
+  private static ArrayList<ACL> createACL(ZooKeeperWatcher zkw, String node) {
+    if (isSecureZooKeeper(zkw.getConfiguration())) {
+      // Certain znodes must be readable by non-authenticated clients
+      if ((node.equals(zkw.rootServerZNode) == true) ||
+          (node.equals(zkw.masterAddressZNode) == true) ||
+          (node.equals(zkw.clusterIdZNode) == true)) {
+        return ZooKeeperWatcher.CREATOR_ALL_AND_WORLD_READABLE;
+      }
+      return Ids.CREATOR_ALL_ACL;
+    } else {
+      return Ids.OPEN_ACL_UNSAFE;
+    }
+  }
+
+  public static void waitForZKConnectionIfAuthenticating(ZooKeeperWatcher zkw)
+      throws InterruptedException {
+    if (isSecureZooKeeper(zkw.getConfiguration())) {
+       LOG.debug("Waiting for ZooKeeperWatcher to authenticate");
+       zkw.saslLatch.await();
+       LOG.debug("Done waiting.");
+    }
+  }
+
   //
   // Node creation
   //
@@ -722,7 +757,8 @@ public class ZKUtil {
       String znode, byte [] data)
   throws KeeperException {
     try {
-      zkw.getRecoverableZooKeeper().create(znode, data, Ids.OPEN_ACL_UNSAFE,
+      waitForZKConnectionIfAuthenticating(zkw);
+      zkw.getRecoverableZooKeeper().create(znode, data, createACL(zkw, znode),
           CreateMode.EPHEMERAL);
     } catch (KeeperException.NodeExistsException nee) {
       if(!watchAndCheckExists(zkw, znode)) {
@@ -761,7 +797,8 @@ public class ZKUtil {
       ZooKeeperWatcher zkw, String znode, byte [] data)
   throws KeeperException {
     try {
-      zkw.getRecoverableZooKeeper().create(znode, data, Ids.OPEN_ACL_UNSAFE,
+      waitForZKConnectionIfAuthenticating(zkw);
+      zkw.getRecoverableZooKeeper().create(znode, data, createACL(zkw, znode),
           CreateMode.PERSISTENT);
     } catch (KeeperException.NodeExistsException nee) {
       try {
@@ -798,7 +835,8 @@ public class ZKUtil {
       String znode, byte [] data)
   throws KeeperException, KeeperException.NodeExistsException {
     try {
-      zkw.getRecoverableZooKeeper().create(znode, data, Ids.OPEN_ACL_UNSAFE,
+      waitForZKConnectionIfAuthenticating(zkw);
+      zkw.getRecoverableZooKeeper().create(znode, data, createACL(zkw, znode),
           CreateMode.PERSISTENT);
       return zkw.getRecoverableZooKeeper().exists(znode, zkw).getVersion();
     } catch (InterruptedException e) {
@@ -825,8 +863,13 @@ public class ZKUtil {
   public static void asyncCreate(ZooKeeperWatcher zkw,
       String znode, byte [] data, final AsyncCallback.StringCallback cb,
       final Object ctx) {
-    zkw.getRecoverableZooKeeper().getZooKeeper().create(znode, data, Ids.OPEN_ACL_UNSAFE,
+    try {
-       CreateMode.PERSISTENT, cb, ctx);
+      waitForZKConnectionIfAuthenticating(zkw);
+      zkw.getRecoverableZooKeeper().getZooKeeper().create(znode, data,
+          createACL(zkw, znode), CreateMode.PERSISTENT, cb, ctx);
+    } catch (InterruptedException e) {
+      zkw.interruptedException(e);
+    }
   }
 
   /**
@@ -844,8 +887,9 @@ public class ZKUtil {
   throws KeeperException {
     try {
       RecoverableZooKeeper zk = zkw.getRecoverableZooKeeper();
+      waitForZKConnectionIfAuthenticating(zkw);
       if (zk.exists(znode, false) == null) {
-        zk.create(znode, new byte[0], Ids.OPEN_ACL_UNSAFE,
+        zk.create(znode, new byte[0], createACL(zkw,znode),
             CreateMode.PERSISTENT);
       }
     } catch(KeeperException.NodeExistsException nee) {
@@ -881,7 +925,8 @@ public class ZKUtil {
       if(znode == null) {
         return;
       }
-      zkw.getRecoverableZooKeeper().create(znode, new byte[0], Ids.OPEN_ACL_UNSAFE,
+      waitForZKConnectionIfAuthenticating(zkw);
+      zkw.getRecoverableZooKeeper().create(znode, new byte[0], createACL(zkw, znode),
           CreateMode.PERSISTENT);
     } catch(KeeperException.NodeExistsException nee) {
       return;

src/main/java/org/apache/hadoop/hbase/zookeeper/ZooKeeperWatcher.java

@@ -20,8 +20,12 @@
 package org.apache.hadoop.hbase.zookeeper;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -33,6 +37,8 @@ import org.apache.hadoop.hbase.util.Threads;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.data.ACL;
 
 /**
  * Acts as the single ZooKeeper Watcher.  One instance of this is instantiated
@@ -48,7 +54,7 @@ import org.apache.zookeeper.Watcher;
 public class ZooKeeperWatcher implements Watcher, Abortable {
   private static final Log LOG = LogFactory.getLog(ZooKeeperWatcher.class);
 
-  // Identifiier for this watcher (for logging only).  Its made of the prefix
+  // Identifier for this watcher (for logging only).  It is made of the prefix
   // passed on construction and the zookeeper sessionid.
   private String identifier;
 
@@ -65,6 +71,13 @@ public class ZooKeeperWatcher implements Watcher, Abortable {
   private final List<ZooKeeperListener> listeners =
     new CopyOnWriteArrayList<ZooKeeperListener>();
 
+  // Used by ZKUtil:waitForZKConnectionIfAuthenticating to wait for SASL
+  // negotiation to complete
+  public CountDownLatch saslLatch = new CountDownLatch(1);
+
+  // set of unassigned nodes watched
+  private Set<String> unassignedNodes = new HashSet<String>();
+
   // node names
 
   // base znode for this cluster
@@ -88,11 +101,17 @@ public class ZooKeeperWatcher implements Watcher, Abortable {
   // znode used for log splitting work assignment
   public String splitLogZNode;
 
+  // Certain ZooKeeper nodes need to be world-readable
+  public static final ArrayList<ACL> CREATOR_ALL_AND_WORLD_READABLE =
+    new ArrayList<ACL>() { {
+      add(new ACL(ZooDefs.Perms.READ,ZooDefs.Ids.ANYONE_ID_UNSAFE));
+      add(new ACL(ZooDefs.Perms.ALL,ZooDefs.Ids.AUTH_IDS));
+    }};
+
   private final Configuration conf;
 
   private final Exception constructorCaller;
 
-
   /**
    * Instantiate a ZooKeeper connection and watcher.
    * @param descriptor Descriptive string that is added to zookeeper sessionid
@@ -315,17 +334,38 @@ public class ZooKeeperWatcher implements Watcher, Abortable {
         LOG.debug(this.identifier + " connected");
         break;
 
+      case SaslAuthenticated:
+        if (ZKUtil.isSecureZooKeeper(this.conf)) {
+          // We are authenticated, clients can proceed.
+          saslLatch.countDown();
+        }
+        break;
+
+      case AuthFailed:
+        if (ZKUtil.isSecureZooKeeper(this.conf)) {
+          // We could not be authenticated, but clients should proceed anyway.
+          // Only access to znodes that require SASL authentication will be
+          // denied. The client may never need to access them.
+          saslLatch.countDown();
+        }
+        break;
+
       // Abort the server if Disconnected or Expired
-      // TODO: Åny reason to handle these two differently?
       case Disconnected:
         LOG.debug(prefix("Received Disconnected from ZooKeeper, ignoring"));
         break;
 
       case Expired:
+        if (ZKUtil.isSecureZooKeeper(this.conf)) {
+          // We consider Expired equivalent to AuthFailed for this
+          // connection. Authentication is never going to complete. The
+          // client should proceed to do cleanup.
+          saslLatch.countDown();
+        }
         String msg = prefix(this.identifier + " received expired from " +
           "ZooKeeper, aborting");
         // TODO: One thought is to add call to ZooKeeperListener so say,
-        // ZooKeperNodeTracker can zero out its data values.
+        // ZooKeeperNodeTracker can zero out its data values.
         if (this.abortable != null) this.abortable.abort(msg,
             new KeeperException.SessionExpiredException());
         break;
@@ -396,6 +436,10 @@ public class ZooKeeperWatcher implements Watcher, Abortable {
     }
   }
 
+  public Configuration getConfiguration() {
+    return conf;
+  }
+
   @Override
   public void abort(String why, Throwable e) {
     this.abortable.abort(why, e);

src/test/java/org/apache/hadoop/hbase/HBaseTestingUtility.java

@@ -420,7 +420,7 @@ public class HBaseTestingUtility {
       throw new IOException("Cluster already running at " + dir);
     }
     this.passedZkCluster = false;
-    this.zkCluster = new MiniZooKeeperCluster();
+    this.zkCluster = new MiniZooKeeperCluster(this.getConfiguration());
     int clientPort =   this.zkCluster.startup(dir,zooKeeperServerNum);
     this.conf.set(HConstants.ZOOKEEPER_CLIENT_PORT,
       Integer.toString(clientPort));

src/test/java/org/apache/hadoop/hbase/zookeeper/TestZooKeeperACL.java

@@ -0,0 +1,266 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.zookeeper;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.TestZooKeeper;
+import org.apache.hadoop.hbase.zookeeper.ZKUtil;
+import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.data.Stat;
+
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+public class TestZooKeeperACL {
+  private final static Log LOG = LogFactory.getLog(TestZooKeeperACL.class);
+  private final static HBaseTestingUtility TEST_UTIL =
+      new HBaseTestingUtility();
+
+  private static ZooKeeperWatcher zkw;
+  private static boolean secureZKAvailable;
+  
+  @BeforeClass
+  public static void setUpBeforeClass() throws Exception {
+    File saslConfFile = File.createTempFile("tmp", "jaas.conf");
+    FileWriter fwriter = new FileWriter(saslConfFile);
+
+    fwriter.write("" +
+      "Server {\n" +
+        "org.apache.zookeeper.server.auth.DigestLoginModule required\n" +
+        "user_hbase=\"secret\";\n" +
+      "};\n" +
+      "Client {\n" +
+        "org.apache.zookeeper.server.auth.DigestLoginModule required\n" +
+        "username=\"hbase\"\n" +
+        "password=\"secret\";\n" +
+      "};" + "\n");
+    fwriter.close();
+    System.setProperty("java.security.auth.login.config",
+        saslConfFile.getAbsolutePath());
+    System.setProperty("zookeeper.authProvider.1",
+        "org.apache.zookeeper.server.auth.SASLAuthenticationProvider");
+
+    TEST_UTIL.getConfiguration().setBoolean("dfs.support.append", true);
+    TEST_UTIL.getConfiguration().setInt("hbase.zookeeper.property.maxClientCnxns", 1000);
+
+    // If Hadoop is missing HADOOP-7070 the cluster will fail to start due to
+    // the JAAS configuration required by ZK being clobbered by Hadoop 
+    try {
+      TEST_UTIL.startMiniCluster();
+    } catch (IOException e) {
+      LOG.warn("Hadoop is missing HADOOP-7070", e);
+      secureZKAvailable = false;
+      return;
+    }
+    zkw = new ZooKeeperWatcher(
+      new Configuration(TEST_UTIL.getConfiguration()),
+        TestZooKeeper.class.getName(), null);
+    ZKUtil.waitForZKConnectionIfAuthenticating(zkw);
+  }
+
+  /**
+   * @throws java.lang.Exception
+   */
+  @AfterClass
+  public static void tearDownAfterClass() throws Exception {
+    if (!secureZKAvailable) {
+      return;
+    }
+    TEST_UTIL.shutdownMiniCluster();
+  }
+
+  /**
+   * @throws java.lang.Exception
+   */
+  @Before
+  public void setUp() throws Exception {
+    if (!secureZKAvailable) {
+      return;
+    }
+    TEST_UTIL.ensureSomeRegionServersAvailable(2);
+  }
+
+  /**
+   * Create a node and check its ACL. When authentication is enabled on 
+   * Zookeeper, all nodes (except /hbase/root-region-server, /hbase/master
+   * and /hbase/hbaseid) should be created so that only the hbase server user
+   * (master or region server user) that created them can access them, and
+   * this user should have all permissions on this node. For
+   * /hbase/root-region-server, /hbase/master, and /hbase/hbaseid the
+   * permissions should be as above, but should also be world-readable. First
+   * we check the general case of /hbase nodes in the following test, and
+   * then check the subset of world-readable nodes in the three tests after
+   * that.
+   */
+  @Test (timeout=30000)
+  public void testHBaseRootZNodeACL() throws Exception {
+    if (!secureZKAvailable) {
+      return;
+    }
+
+    List<ACL> acls = zkw.getRecoverableZooKeeper().getZooKeeper()
+        .getACL("/hbase", new Stat());
+    assertEquals(acls.size(),1);
+    assertEquals(acls.get(0).getId().getScheme(),"sasl");
+    assertEquals(acls.get(0).getId().getId(),"hbase");
+    assertEquals(acls.get(0).getPerms(), ZooDefs.Perms.ALL);
+  }
+
+  /**
+   * When authentication is enabled on Zookeeper, /hbase/root-region-server
+   * should be created with 2 ACLs: one specifies that the hbase user has
+   * full access to the node; the other, that it is world-readable.
+   */
+  @Test (timeout=30000)
+  public void testHBaseRootRegionServerZNodeACL() throws Exception {
+    if (!secureZKAvailable) {
+      return;
+    }
+
+    List<ACL> acls = zkw.getRecoverableZooKeeper().getZooKeeper()
+        .getACL("/hbase/root-region-server", new Stat());
+    assertEquals(acls.size(),2);
+
+    boolean foundWorldReadableAcl = false;
+    boolean foundHBaseOwnerAcl = false;
+    for(int i = 0; i < 2; i++) {
+      if (acls.get(i).getId().getScheme().equals("world") == true) {
+        assertEquals(acls.get(0).getId().getId(),"anyone");
+        assertEquals(acls.get(0).getPerms(), ZooDefs.Perms.READ);
+        foundWorldReadableAcl = true;
+      }
+      else {
+        if (acls.get(i).getId().getScheme().equals("sasl") == true) {
+          assertEquals(acls.get(1).getId().getId(),"hbase");
+          assertEquals(acls.get(1).getId().getScheme(),"sasl");
+          foundHBaseOwnerAcl = true;
+        } else { // error: should not get here: test fails.
+          assertTrue(false);
+        }
+      }
+    }
+    assertTrue(foundWorldReadableAcl);
+    assertTrue(foundHBaseOwnerAcl);
+  }
+
+  /**
+   * When authentication is enabled on Zookeeper, /hbase/master should be
+   * created with 2 ACLs: one specifies that the hbase user has full access
+   * to the node; the other, that it is world-readable.
+   */
+  @Test (timeout=30000)
+  public void testHBaseMasterServerZNodeACL() throws Exception {
+    if (!secureZKAvailable) {
+      return;
+    }
+
+    List<ACL> acls = zkw.getRecoverableZooKeeper().getZooKeeper()
+        .getACL("/hbase/master", new Stat());
+    assertEquals(acls.size(),2);
+
+    boolean foundWorldReadableAcl = false;
+    boolean foundHBaseOwnerAcl = false;
+    for(int i = 0; i < 2; i++) {
+      if (acls.get(i).getId().getScheme().equals("world") == true) {
+        assertEquals(acls.get(0).getId().getId(),"anyone");
+        assertEquals(acls.get(0).getPerms(), ZooDefs.Perms.READ);
+        foundWorldReadableAcl = true;
+      } else {
+        if (acls.get(i).getId().getScheme().equals("sasl") == true) {
+          assertEquals(acls.get(1).getId().getId(),"hbase");
+          assertEquals(acls.get(1).getId().getScheme(),"sasl");
+          foundHBaseOwnerAcl = true;
+        } else { // error: should not get here: test fails.
+          assertTrue(false);
+        }
+      }
+    }
+    assertTrue(foundWorldReadableAcl);
+    assertTrue(foundHBaseOwnerAcl);
+  }
+
+  /**
+   * When authentication is enabled on Zookeeper, /hbase/hbaseid should be
+   * created with 2 ACLs: one specifies that the hbase user has full access
+   * to the node; the other, that it is world-readable.
+   */
+  @Test (timeout=30000)
+  public void testHBaseIDZNodeACL() throws Exception {
+    if (!secureZKAvailable) {
+      return;
+    }
+
+    List<ACL> acls = zkw.getRecoverableZooKeeper().getZooKeeper()
+        .getACL("/hbase/hbaseid", new Stat());
+    assertEquals(acls.size(),2);
+
+    boolean foundWorldReadableAcl = false;
+    boolean foundHBaseOwnerAcl = false;
+    for(int i = 0; i < 2; i++) {
+      if (acls.get(i).getId().getScheme().equals("world") == true) {
+        assertEquals(acls.get(0).getId().getId(),"anyone");
+        assertEquals(acls.get(0).getPerms(), ZooDefs.Perms.READ);
+        foundWorldReadableAcl = true;
+      } else {
+        if (acls.get(i).getId().getScheme().equals("sasl") == true) {
+          assertEquals(acls.get(1).getId().getId(),"hbase");
+          assertEquals(acls.get(1).getId().getScheme(),"sasl");
+          foundHBaseOwnerAcl = true;
+        } else { // error: should not get here: test fails.
+          assertTrue(false);
+        }
+      }
+    }
+    assertTrue(foundWorldReadableAcl);
+    assertTrue(foundHBaseOwnerAcl);
+  }
+
+  /**
+   * Finally, we check the ACLs of a node outside of the /hbase hierarchy and
+   * verify that its ACL is simply 'hbase:Perms.ALL'.
+   */
+  @Test
+  public void testOutsideHBaseNodeACL() throws Exception {
+    if (!secureZKAvailable) {
+      return;
+    }
+
+    ZKUtil.createWithParents(zkw, "/testACLNode");
+    List<ACL> acls = zkw.getRecoverableZooKeeper().getZooKeeper()
+        .getACL("/testACLNode", new Stat());
+    assertEquals(acls.size(),1);
+    assertEquals(acls.get(0).getId().getScheme(),"sasl");
+    assertEquals(acls.get(0).getId().getId(),"hbase");
+    assertEquals(acls.get(0).getPerms(), ZooDefs.Perms.ALL);
+  }
+}