ZKAclSet

ZKAclSet.java

/**
 *
 * 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.
 */

import java.util.ArrayList;
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.conf.Configured;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
import org.apache.hadoop.hbase.zookeeper.ZKUtil;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooDefs.Perms;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Id;
import org.apache.zookeeper.data.Stat;

/**
 * You may add the jaas.conf option
 *    -Djava.security.auth.login.config=/PATH/jaas.conf
 *
 * You may also specify -D to set options
 *    "hbase.zookeeper.quorum"    (it should be in hbase-site.xml)
 *    "zookeeper.znode.parent"    (it should be in hbase-site.xml)
 */
@InterfaceAudience.Private
public class ZkAclSet extends Configured implements Tool {
  private static final Log LOG = LogFactory.getLog(ZkAclSet.class);

  private static final int ZK_SESSION_TIMEOUT_DEFAULT = 5 * 1000;

  private String baseZNode;
  private String metaServerZNode;
  private String rsZNode;
  private String drainingZNode;
  private String masterAddressZNode;
  private String backupMasterAddressesZNode;
  private String tableZNode;
  private String clusterIdZNode;

  private ArrayList<ACL> createACL(String node) {
    if (!node.startsWith(baseZNode)) {
      return Ids.OPEN_ACL_UNSAFE;
    }

    String superUser = getConf().get("hbase.superuser");
    ArrayList<ACL> acls = new ArrayList<ACL>();
    // add permission to hbase supper user
    if (superUser != null) {
      acls.add(new ACL(Perms.ALL, new Id("auth", superUser)));
    }
    // Certain znodes are accessed directly by the client,
    // so they must be readable by non-authenticated clients
    if ((node.equals(baseZNode) == true) ||
        (node.startsWith(metaServerZNode) == true) ||
        (node.equals(masterAddressZNode) == true) ||
        (node.equals(clusterIdZNode) == true) ||
        (node.equals(rsZNode) == true) ||
        (node.equals(backupMasterAddressesZNode) == true) ||
        (node.startsWith(tableZNode) == true)) {
      acls.addAll(Ids.CREATOR_ALL_ACL);
      acls.addAll(Ids.READ_ACL_UNSAFE);
    } else {
      acls.addAll(Ids.CREATOR_ALL_ACL);
    }
    return acls;
  }

  private void resetAcls(final ZooKeeper zk, final String znode)
      throws Exception {
    List<String> children = zk.getChildren(znode, false);
    if (children != null) {
      for (String child: children) {
        resetAcls(zk, znode + '/' + child);
      }
    }
    LOG.info(" - reset acl for " + znode);
    zk.setACL(znode, createACL(znode), -1);
  }

  private void resetAcls(final String quorumServers, final int zkTimeout, final String znode)
      throws Exception {
    ZooKeeper zk = new ZooKeeper(quorumServers, zkTimeout, new Watcher() {
      @Override
      public void process(WatchedEvent event) {
        LOG.info("Received ZooKeeper Event, " +
            "type=" + event.getType() + ", " +
            "state=" + event.getState() + ", " +
            "path=" + event.getPath());
      }
    });
    try {
      resetAcls(zk, znode);
    } finally {
      zk.close();
    }
  }

  private void resetHBaseAcls(final Configuration conf) throws Exception {
    String quorumServers = conf.get("hbase.zookeeper.quorum", HConstants.LOCALHOST);
    int sessionTimeout = conf.getInt("zookeeper.session.timeout", ZK_SESSION_TIMEOUT_DEFAULT);
    String znode = conf.get("zookeeper.znode.parent", HConstants.DEFAULT_ZOOKEEPER_ZNODE_PARENT);
    if (quorumServers == null) {
      LOG.error("Unable to load hbase.zookeeper.quorum (try with: -conf hbase-site.xml)");
      return;
    }

    LOG.info("Set HBase ACLs for " + quorumServers + " " + znode);
    resetAcls(quorumServers, sessionTimeout, znode);
  }


  @Override
  public int run(String[] args) throws Exception {
    Configuration conf = getConf();
    baseZNode = conf.get(HConstants.ZOOKEEPER_ZNODE_PARENT,
        HConstants.DEFAULT_ZOOKEEPER_ZNODE_PARENT);
    metaServerZNode = ZKUtil.joinZNode(baseZNode,
        conf.get("zookeeper.znode.metaserver", "meta-region-server") + "-");
    rsZNode = ZKUtil.joinZNode(baseZNode,
        conf.get("zookeeper.znode.rs", "rs"));
    drainingZNode = ZKUtil.joinZNode(baseZNode,
        conf.get("zookeeper.znode.draining.rs", "draining"));
    masterAddressZNode = ZKUtil.joinZNode(baseZNode,
        conf.get("zookeeper.znode.master", "master"));
    backupMasterAddressesZNode = ZKUtil.joinZNode(baseZNode,
        conf.get("zookeeper.znode.backup.masters", "backup-masters"));
    tableZNode = ZKUtil.joinZNode(baseZNode,
        conf.get("zookeeper.znode.tableEnableDisable", "table"));
    clusterIdZNode = ZKUtil.joinZNode(baseZNode,
        conf.get("zookeeper.znode.clusterId", "hbaseid"));

    resetHBaseAcls(conf);
    return(0);
  }

  public static void main(String[] args) throws Exception {
    System.exit(ToolRunner.run(new Configuration(), new ZkAclSet(), args));
  }
}