HBASE-19483 Add proper privilege check for rsgroup commands

Signed-off-by: tedyu <yuzhihong@gmail.com>

hbase-it/src/test/java/org/apache/hadoop/hbase/IntegrationTestIngestWithVisibilityLabels.java

@@ -23,11 +23,10 @@ import java.util.Arrays;
 import java.util.List;
 
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.io.hfile.HFile;
 import org.apache.hadoop.hbase.security.User;
 import org.apache.hadoop.hbase.security.visibility.LoadTestDataGeneratorWithVisibilityLabels;
 import org.apache.hadoop.hbase.security.visibility.VisibilityClient;
-import org.apache.hadoop.hbase.security.visibility.VisibilityController;
+import org.apache.hadoop.hbase.security.visibility.VisibilityTestUtil;
 import org.apache.hadoop.hbase.testclassification.IntegrationTests;
 import org.apache.hadoop.hbase.util.LoadTestTool;
 import org.junit.experimental.categories.Category;
@@ -76,9 +75,7 @@ public class IntegrationTestIngestWithVisibilityLabels extends IntegrationTestIn
   public void setUpCluster() throws Exception {
     util = getTestingUtil(null);
     Configuration conf = util.getConfiguration();
-    conf.setInt(HFile.FORMAT_VERSION_KEY, 3);
+    VisibilityTestUtil.enableVisiblityLabels(conf);
-    conf.set("hbase.coprocessor.master.classes", VisibilityController.class.getName());
-    conf.set("hbase.coprocessor.region.classes", VisibilityController.class.getName());
     conf.set("hbase.superuser", "admin," + User.getCurrent().getName());
     super.setUpCluster();
     addLabels();

hbase-it/src/test/java/org/apache/hadoop/hbase/test/IntegrationTestBigLinkedListWithVisibility.java

@@ -47,7 +47,6 @@ import org.apache.hadoop.hbase.client.Put;
 import org.apache.hadoop.hbase.client.Result;
 import org.apache.hadoop.hbase.client.Scan;
 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
-import org.apache.hadoop.hbase.io.hfile.HFile;
 import org.apache.hadoop.hbase.log.HBaseMarkers;
 import org.apache.hadoop.hbase.mapreduce.Import;
 import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
@@ -57,7 +56,7 @@ import org.apache.hadoop.hbase.security.access.Permission;
 import org.apache.hadoop.hbase.security.visibility.Authorizations;
 import org.apache.hadoop.hbase.security.visibility.CellVisibility;
 import org.apache.hadoop.hbase.security.visibility.VisibilityClient;
-import org.apache.hadoop.hbase.security.visibility.VisibilityController;
+import org.apache.hadoop.hbase.security.visibility.VisibilityTestUtil;
 import org.apache.hadoop.hbase.testclassification.IntegrationTests;
 import org.apache.hadoop.hbase.util.AbstractHBaseTool;
 import org.apache.hadoop.hbase.util.Bytes;
@@ -373,9 +372,7 @@ public class IntegrationTestBigLinkedListWithVisibility extends IntegrationTestB
   public void setUpCluster() throws Exception {
     util = getTestingUtil(null);
     Configuration conf = util.getConfiguration();
-    conf.setInt(HFile.FORMAT_VERSION_KEY, 3);
+    VisibilityTestUtil.enableVisiblityLabels(conf);
-    conf.set("hbase.coprocessor.master.classes", VisibilityController.class.getName());
-    conf.set("hbase.coprocessor.region.classes", VisibilityController.class.getName());
     conf.set("hbase.superuser", User.getCurrent().getName());
     conf.setBoolean("dfs.permissions", false);
     USER = User.createUserForTesting(conf, userName, new String[] {});

hbase-it/src/test/java/org/apache/hadoop/hbase/test/IntegrationTestWithCellVisibilityLoadAndVerify.java

@@ -40,7 +40,6 @@ import org.apache.hadoop.hbase.client.Result;
 import org.apache.hadoop.hbase.client.Scan;
 import org.apache.hadoop.hbase.client.ScannerCallable;
 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
-import org.apache.hadoop.hbase.io.hfile.HFile;
 import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
 import org.apache.hadoop.hbase.mapreduce.TableMapper;
 import org.apache.hadoop.hbase.mapreduce.TableRecordReaderImpl;
@@ -48,7 +47,7 @@ import org.apache.hadoop.hbase.security.User;
 import org.apache.hadoop.hbase.security.visibility.Authorizations;
 import org.apache.hadoop.hbase.security.visibility.CellVisibility;
 import org.apache.hadoop.hbase.security.visibility.VisibilityClient;
-import org.apache.hadoop.hbase.security.visibility.VisibilityController;
+import org.apache.hadoop.hbase.security.visibility.VisibilityTestUtil;
 import org.apache.hadoop.hbase.testclassification.IntegrationTests;
 import org.apache.hadoop.hbase.util.AbstractHBaseTool;
 import org.apache.hadoop.hbase.util.Bytes;
@@ -116,9 +115,7 @@ public class IntegrationTestWithCellVisibilityLoadAndVerify extends IntegrationT
   public void setUpCluster() throws Exception {
     util = getTestingUtil(null);
     Configuration conf = util.getConfiguration();
-    conf.setInt(HFile.FORMAT_VERSION_KEY, 3);
+    VisibilityTestUtil.enableVisiblityLabels(conf);
-    conf.set("hbase.coprocessor.master.classes", VisibilityController.class.getName());
-    conf.set("hbase.coprocessor.region.classes", VisibilityController.class.getName());
     conf.set("hbase.superuser", User.getCurrent().getName());
     conf.setBoolean("dfs.permissions", false);
     super.setUpCluster();

hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestImportTSVWithVisibilityLabels.java

@@ -58,7 +58,7 @@ import org.apache.hadoop.hbase.security.visibility.ScanLabelGenerator;
 import org.apache.hadoop.hbase.security.visibility.SimpleScanLabelGenerator;
 import org.apache.hadoop.hbase.security.visibility.VisibilityClient;
 import org.apache.hadoop.hbase.security.visibility.VisibilityConstants;
-import org.apache.hadoop.hbase.security.visibility.VisibilityController;
+import org.apache.hadoop.hbase.security.visibility.VisibilityTestUtil;
 import org.apache.hadoop.hbase.security.visibility.VisibilityUtils;
 import org.apache.hadoop.hbase.testclassification.LargeTests;
 import org.apache.hadoop.hbase.testclassification.MapReduceTests;
@@ -121,9 +121,7 @@ public class TestImportTSVWithVisibilityLabels implements Configurable {
     conf = util.getConfiguration();
     SUPERUSER = User.createUserForTesting(conf, "admin", new String[] { "supergroup" });
     conf.set("hbase.superuser", "admin,"+User.getCurrent().getName());
-    conf.setInt("hfile.format.version", 3);
+    VisibilityTestUtil.enableVisiblityLabels(conf);
-    conf.set("hbase.coprocessor.master.classes", VisibilityController.class.getName());
-    conf.set("hbase.coprocessor.region.classes", VisibilityController.class.getName());
     conf.setClass(VisibilityUtils.VISIBILITY_LABEL_GENERATOR_CLASS, SimpleScanLabelGenerator.class,
         ScanLabelGenerator.class);
     util.startMiniCluster();

hbase-rest/src/test/java/org/apache/hadoop/hbase/rest/TestScannersWithLabels.java

@@ -43,7 +43,7 @@ import org.apache.hadoop.hbase.security.visibility.ScanLabelGenerator;
 import org.apache.hadoop.hbase.security.visibility.SimpleScanLabelGenerator;
 import org.apache.hadoop.hbase.security.visibility.VisibilityClient;
 import org.apache.hadoop.hbase.security.visibility.VisibilityConstants;
-import org.apache.hadoop.hbase.security.visibility.VisibilityController;
+import org.apache.hadoop.hbase.security.visibility.VisibilityTestUtil;
 import org.apache.hadoop.hbase.security.visibility.VisibilityUtils;
 import org.apache.hadoop.hbase.testclassification.MediumTests;
 import org.apache.hadoop.hbase.testclassification.RestTests;
@@ -130,10 +130,8 @@ public class TestScannersWithLabels {
     conf = TEST_UTIL.getConfiguration();
     conf.setClass(VisibilityUtils.VISIBILITY_LABEL_GENERATOR_CLASS,
         SimpleScanLabelGenerator.class, ScanLabelGenerator.class);
-    conf.setInt("hfile.format.version", 3);
     conf.set("hbase.superuser", SUPERUSER.getShortName());
-    conf.set("hbase.coprocessor.master.classes", VisibilityController.class.getName());
+    VisibilityTestUtil.enableVisiblityLabels(conf);
-    conf.set("hbase.coprocessor.region.classes", VisibilityController.class.getName());
     TEST_UTIL.startMiniCluster(1);
     // Wait for the labels table to become available
     TEST_UTIL.waitTableEnabled(VisibilityConstants.LABELS_TABLE_NAME.getName(), 50000);

hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupAdminEndpoint.java

@@ -46,6 +46,7 @@ import org.apache.hadoop.hbase.coprocessor.MasterCoprocessorEnvironment;
 import org.apache.hadoop.hbase.coprocessor.MasterObserver;
 import org.apache.hadoop.hbase.coprocessor.ObserverContext;
 import org.apache.hadoop.hbase.ipc.CoprocessorRpcUtils;
+import org.apache.hadoop.hbase.ipc.RpcServer;
 import org.apache.hadoop.hbase.master.MasterServices;
 import org.apache.hadoop.hbase.net.Address;
 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
@@ -74,6 +75,12 @@ import org.apache.hadoop.hbase.protobuf.generated.RSGroupAdminProtos.RemoveRSGro
 import org.apache.hadoop.hbase.protobuf.generated.RSGroupAdminProtos.RemoveRSGroupResponse;
 import org.apache.hadoop.hbase.protobuf.generated.RSGroupAdminProtos.RemoveServersRequest;
 import org.apache.hadoop.hbase.protobuf.generated.RSGroupAdminProtos.RemoveServersResponse;
+import org.apache.hadoop.hbase.security.User;
+import org.apache.hadoop.hbase.security.UserProvider;
+import org.apache.hadoop.hbase.security.access.AccessChecker;
+import org.apache.hadoop.hbase.security.access.Permission.Action;
+import org.apache.hadoop.hbase.security.access.TableAuthManager;
+import org.apache.hadoop.hbase.zookeeper.ZKWatcher;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -92,12 +99,17 @@ public class RSGroupAdminEndpoint implements MasterCoprocessor, MasterObserver {
   private RSGroupInfoManager groupInfoManager;
   private RSGroupAdminServer groupAdminServer;
   private final RSGroupAdminService groupAdminService = new RSGroupAdminServiceImpl();
+  private AccessChecker accessChecker;
+
+  /** Provider for mapping principal names to Users */
+  private UserProvider userProvider;
 
   @Override
   public void start(CoprocessorEnvironment env) throws IOException {
     if (!(env instanceof HasMasterServices)) {
       throw new IOException("Does not implement HMasterServices");
     }
+
     master = ((HasMasterServices)env).getMasterServices();
     groupInfoManager = RSGroupInfoManagerImpl.getInstance(master);
     groupAdminServer = new RSGroupAdminServer(master, groupInfoManager);
@@ -106,6 +118,16 @@ public class RSGroupAdminEndpoint implements MasterCoprocessor, MasterObserver {
     if (!RSGroupableBalancer.class.isAssignableFrom(clazz)) {
       throw new IOException("Configured balancer does not support RegionServer groups.");
     }
+    ZKWatcher zk = ((HasMasterServices)env).getMasterServices().getZooKeeper();
+    accessChecker = new AccessChecker(env.getConfiguration(), zk);
+
+    // set the user-provider.
+    this.userProvider = UserProvider.instantiate(env.getConfiguration());
+  }
+
+  @Override
+  public void stop(CoprocessorEnvironment env) {
+    TableAuthManager.release(accessChecker.getAuthManager());
   }
 
   @Override
@@ -137,6 +159,7 @@ public class RSGroupAdminEndpoint implements MasterCoprocessor, MasterObserver {
       LOG.info(master.getClientIdAuditPrefix() + " initiates rsgroup info retrieval, group="
               + groupName);
       try {
+        checkPermission("getRSGroupInfo");
         RSGroupInfo rsGroupInfo = groupAdminServer.getRSGroupInfo(groupName);
         if (rsGroupInfo != null) {
           builder.setRSGroupInfo(RSGroupProtobufUtil.toProtoGroupInfo(rsGroupInfo));
@@ -151,10 +174,11 @@ public class RSGroupAdminEndpoint implements MasterCoprocessor, MasterObserver {
     public void getRSGroupInfoOfTable(RpcController controller,
         GetRSGroupInfoOfTableRequest request, RpcCallback<GetRSGroupInfoOfTableResponse> done) {
       GetRSGroupInfoOfTableResponse.Builder builder = GetRSGroupInfoOfTableResponse.newBuilder();
-      try {
       TableName tableName = ProtobufUtil.toTableName(request.getTableName());
       LOG.info(master.getClientIdAuditPrefix() + " initiates rsgroup info retrieval, table="
           + tableName);
+      try {
+        checkPermission("getRSGroupInfoOfTable");
         RSGroupInfo RSGroupInfo = groupAdminServer.getRSGroupInfoOfTable(tableName);
         if (RSGroupInfo != null) {
           builder.setRSGroupInfo(RSGroupProtobufUtil.toProtoGroupInfo(RSGroupInfo));
@@ -169,13 +193,14 @@ public class RSGroupAdminEndpoint implements MasterCoprocessor, MasterObserver {
     public void moveServers(RpcController controller, MoveServersRequest request,
         RpcCallback<MoveServersResponse> done) {
       MoveServersResponse.Builder builder = MoveServersResponse.newBuilder();
-      try {
       Set<Address> hostPorts = Sets.newHashSet();
       for (HBaseProtos.ServerName el : request.getServersList()) {
         hostPorts.add(Address.fromParts(el.getHostName(), el.getPort()));
       }
       LOG.info(master.getClientIdAuditPrefix() + " move servers " + hostPorts +" to rsgroup "
           + request.getTargetGroup());
+      try {
+        checkPermission("moveServers");
         groupAdminServer.moveServers(hostPorts, request.getTargetGroup());
       } catch (IOException e) {
         CoprocessorRpcUtils.setControllerException(controller, e);
@@ -187,13 +212,14 @@ public class RSGroupAdminEndpoint implements MasterCoprocessor, MasterObserver {
     public void moveTables(RpcController controller, MoveTablesRequest request,
         RpcCallback<MoveTablesResponse> done) {
       MoveTablesResponse.Builder builder = MoveTablesResponse.newBuilder();
-      try {
       Set<TableName> tables = new HashSet<>(request.getTableNameList().size());
       for (HBaseProtos.TableName tableName : request.getTableNameList()) {
         tables.add(ProtobufUtil.toTableName(tableName));
       }
       LOG.info(master.getClientIdAuditPrefix() + " move tables " + tables +" to rsgroup "
           + request.getTargetGroup());
+      try {
+        checkPermission("moveTables");
         groupAdminServer.moveTables(tables, request.getTargetGroup());
       } catch (IOException e) {
         CoprocessorRpcUtils.setControllerException(controller, e);
@@ -207,6 +233,7 @@ public class RSGroupAdminEndpoint implements MasterCoprocessor, MasterObserver {
       AddRSGroupResponse.Builder builder = AddRSGroupResponse.newBuilder();
       LOG.info(master.getClientIdAuditPrefix() + " add rsgroup " + request.getRSGroupName());
       try {
+        checkPermission("addRSGroup");
         groupAdminServer.addRSGroup(request.getRSGroupName());
       } catch (IOException e) {
         CoprocessorRpcUtils.setControllerException(controller, e);
@@ -221,6 +248,7 @@ public class RSGroupAdminEndpoint implements MasterCoprocessor, MasterObserver {
           RemoveRSGroupResponse.newBuilder();
       LOG.info(master.getClientIdAuditPrefix() + " remove rsgroup " + request.getRSGroupName());
       try {
+        checkPermission("removeRSGroup");
         groupAdminServer.removeRSGroup(request.getRSGroupName());
       } catch (IOException e) {
         CoprocessorRpcUtils.setControllerException(controller, e);
@@ -235,6 +263,7 @@ public class RSGroupAdminEndpoint implements MasterCoprocessor, MasterObserver {
       LOG.info(master.getClientIdAuditPrefix() + " balance rsgroup, group=" +
               request.getRSGroupName());
       try {
+        checkPermission("balanceRSGroup");
         builder.setBalanceRan(groupAdminServer.balanceRSGroup(request.getRSGroupName()));
       } catch (IOException e) {
         CoprocessorRpcUtils.setControllerException(controller, e);
@@ -249,6 +278,7 @@ public class RSGroupAdminEndpoint implements MasterCoprocessor, MasterObserver {
       ListRSGroupInfosResponse.Builder builder = ListRSGroupInfosResponse.newBuilder();
       LOG.info(master.getClientIdAuditPrefix() + " list rsgroup");
       try {
+        checkPermission("listRSGroup");
         for (RSGroupInfo RSGroupInfo : groupAdminServer.listRSGroups()) {
           builder.addRSGroupInfo(RSGroupProtobufUtil.toProtoGroupInfo(RSGroupInfo));
         }
@@ -262,11 +292,12 @@ public class RSGroupAdminEndpoint implements MasterCoprocessor, MasterObserver {
     public void getRSGroupInfoOfServer(RpcController controller,
         GetRSGroupInfoOfServerRequest request, RpcCallback<GetRSGroupInfoOfServerResponse> done) {
       GetRSGroupInfoOfServerResponse.Builder builder = GetRSGroupInfoOfServerResponse.newBuilder();
-      try {
       Address hp = Address.fromParts(request.getServer().getHostName(),
           request.getServer().getPort());
-        LOG.info(master.getClientIdAuditPrefix() + " initiates rsgroup info retrieval, server=" +
+      LOG.info(master.getClientIdAuditPrefix() + " initiates rsgroup info retrieval, server="
-                hp);
+          + hp);
+      try {
+        checkPermission("getRSGroupInfoOfServer");
         RSGroupInfo RSGroupInfo = groupAdminServer.getRSGroupOfServer(hp);
         if (RSGroupInfo != null) {
           builder.setRSGroupInfo(RSGroupProtobufUtil.toProtoGroupInfo(RSGroupInfo));
@@ -281,7 +312,6 @@ public class RSGroupAdminEndpoint implements MasterCoprocessor, MasterObserver {
     public void moveServersAndTables(RpcController controller,
         MoveServersAndTablesRequest request, RpcCallback<MoveServersAndTablesResponse> done) {
       MoveServersAndTablesResponse.Builder builder = MoveServersAndTablesResponse.newBuilder();
-      try {
       Set<Address> hostPorts = Sets.newHashSet();
       for (HBaseProtos.ServerName el : request.getServersList()) {
         hostPorts.add(Address.fromParts(el.getHostName(), el.getPort()));
@@ -292,6 +322,8 @@ public class RSGroupAdminEndpoint implements MasterCoprocessor, MasterObserver {
       }
       LOG.info(master.getClientIdAuditPrefix() + " move servers " + hostPorts
           + " and tables " + tables + " to rsgroup" + request.getTargetGroup());
+      try {
+        checkPermission("moveServersAndTables");
         groupAdminServer.moveServersAndTables(hostPorts, tables, request.getTargetGroup());
       } catch (IOException e) {
         CoprocessorRpcUtils.setControllerException(controller, e);
@@ -305,13 +337,14 @@ public class RSGroupAdminEndpoint implements MasterCoprocessor, MasterObserver {
         RpcCallback<RemoveServersResponse> done) {
       RemoveServersResponse.Builder builder =
           RemoveServersResponse.newBuilder();
-      try {
       Set<Address> servers = Sets.newHashSet();
       for (HBaseProtos.ServerName el : request.getServersList()) {
         servers.add(Address.fromParts(el.getHostName(), el.getPort()));
       }
       LOG.info(master.getClientIdAuditPrefix()
           + " remove decommissioned servers from rsgroup: " + servers);
+      try {
+        checkPermission("removeServers");
         groupAdminServer.removeServers(servers);
       } catch (IOException e) {
         CoprocessorRpcUtils.setControllerException(controller, e);
@@ -397,5 +430,21 @@ public class RSGroupAdminEndpoint implements MasterCoprocessor, MasterObserver {
     groupAdminServer.removeServers(clearedServer);
   }
 
-  /////////////////////////////////////////////////////////////////////////////
+  public void checkPermission(String request) throws IOException {
+    accessChecker.requirePermission(getActiveUser(), request, Action.ADMIN);
+  }
+
+  /**
+   * Returns the active user to which authorization checks should be applied.
+   * If we are in the context of an RPC call, the remote user is used,
+   * otherwise the currently logged in user is used.
+   */
+  private User getActiveUser() throws IOException {
+    // for non-rpc handling, fallback to system user
+    Optional<User> optionalUser = RpcServer.getRequestUser();
+    if (optionalUser.isPresent()) {
+      return optionalUser.get();
+    }
+    return userProvider.getCurrent();
+  }
 }

hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/rsgroup/TestRSGroupsWithACL.java

@@ -0,0 +1,378 @@
+
+/*
+ * 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.rsgroup;
+
+import static org.apache.hadoop.hbase.AuthUtil.toGroupEntry;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.TableNotFoundException;
+import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
+import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
+import org.apache.hadoop.hbase.master.MasterCoprocessorHost;
+import org.apache.hadoop.hbase.security.User;
+import org.apache.hadoop.hbase.security.access.AccessControlClient;
+import org.apache.hadoop.hbase.security.access.AccessControlLists;
+import org.apache.hadoop.hbase.security.access.Permission;
+import org.apache.hadoop.hbase.security.access.SecureTestUtil;
+import org.apache.hadoop.hbase.security.access.SecureTestUtil.AccessTestAction;
+import org.apache.hadoop.hbase.security.access.TableAuthManager;
+import org.apache.hadoop.hbase.testclassification.SecurityTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TestName;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@Category({SecurityTests.class})
+public class TestRSGroupsWithACL {
+  private static final Logger LOG = LoggerFactory.getLogger(TestRSGroupsWithACL.class);
+  private static TableName TEST_TABLE = TableName.valueOf("testtable1");
+  private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
+  private static Configuration conf;
+
+  private static Connection systemUserConnection;
+  // user with all permissions
+  private static User SUPERUSER;
+  // user granted with all global permission
+  private static User USER_ADMIN;
+  // user with rw permissions on column family.
+  private static User USER_RW;
+  // user with read-only permissions
+  private static User USER_RO;
+  // user is table owner. will have all permissions on table
+  private static User USER_OWNER;
+  // user with create table permissions alone
+  private static User USER_CREATE;
+  // user with no permissions
+  private static User USER_NONE;
+  // user with admin rights on the column family
+  private static User USER_ADMIN_CF;
+
+  private static final String GROUP_ADMIN = "group_admin";
+  private static final String GROUP_CREATE = "group_create";
+  private static final String GROUP_READ = "group_read";
+  private static final String GROUP_WRITE = "group_write";
+
+  private static User USER_GROUP_ADMIN;
+  private static User USER_GROUP_CREATE;
+  private static User USER_GROUP_READ;
+  private static User USER_GROUP_WRITE;
+
+  private static byte[] TEST_FAMILY = Bytes.toBytes("f1");
+
+  private static RSGroupAdminEndpoint rsGroupAdminEndpoint;
+
+  @Rule
+  public TestName name = new TestName();
+
+  @BeforeClass
+  public static void setupBeforeClass() throws Exception {
+    // setup configuration
+    conf = TEST_UTIL.getConfiguration();
+    conf.set(HConstants.HBASE_MASTER_LOADBALANCER_CLASS,
+        RSGroupBasedLoadBalancer.class.getName());
+    // Enable security
+    SecureTestUtil.enableSecurity(conf);
+    // Verify enableSecurity sets up what we require
+    SecureTestUtil.verifyConfiguration(conf);
+
+    configureRSGroupAdminEndpoint(conf);
+
+    TEST_UTIL.startMiniCluster();
+    MasterCoprocessorHost cpHost =
+        TEST_UTIL.getMiniHBaseCluster().getMaster().getMasterCoprocessorHost();
+    rsGroupAdminEndpoint = (RSGroupAdminEndpoint)
+        cpHost.findCoprocessor(RSGroupAdminEndpoint.class.getName());
+
+
+    // Wait for the ACL table to become available
+    TEST_UTIL.waitUntilAllRegionsAssigned(AccessControlLists.ACL_TABLE_NAME);
+
+    // create a set of test users
+    SUPERUSER = User.createUserForTesting(conf, "admin", new String[] { "supergroup" });
+    USER_ADMIN = User.createUserForTesting(conf, "admin2", new String[0]);
+    USER_RW = User.createUserForTesting(conf, "rwuser", new String[0]);
+    USER_RO = User.createUserForTesting(conf, "rouser", new String[0]);
+    USER_OWNER = User.createUserForTesting(conf, "owner", new String[0]);
+    USER_CREATE = User.createUserForTesting(conf, "tbl_create", new String[0]);
+    USER_NONE = User.createUserForTesting(conf, "nouser", new String[0]);
+    USER_ADMIN_CF = User.createUserForTesting(conf, "col_family_admin", new String[0]);
+
+    USER_GROUP_ADMIN =
+        User.createUserForTesting(conf, "user_group_admin", new String[] { GROUP_ADMIN });
+    USER_GROUP_CREATE =
+        User.createUserForTesting(conf, "user_group_create", new String[] { GROUP_CREATE });
+    USER_GROUP_READ =
+        User.createUserForTesting(conf, "user_group_read", new String[] { GROUP_READ });
+    USER_GROUP_WRITE =
+        User.createUserForTesting(conf, "user_group_write", new String[] { GROUP_WRITE });
+
+    systemUserConnection = TEST_UTIL.getConnection();
+    setUpTableAndUserPermissions();
+  }
+
+  private static void setUpTableAndUserPermissions() throws Exception {
+    TableDescriptorBuilder tableBuilder = TableDescriptorBuilder.newBuilder(TEST_TABLE);
+    ColumnFamilyDescriptorBuilder cfd = ColumnFamilyDescriptorBuilder.newBuilder(TEST_FAMILY);
+    cfd.setMaxVersions(100);
+    tableBuilder.addColumnFamily(cfd.build());
+    tableBuilder.setValue(TableDescriptorBuilder.OWNER, USER_OWNER.getShortName());
+    SecureTestUtil.createTable(TEST_UTIL, tableBuilder.build(),
+        new byte[][] { Bytes.toBytes("s") });
+
+    // Set up initial grants
+    SecureTestUtil.grantGlobal(TEST_UTIL, USER_ADMIN.getShortName(),
+        Permission.Action.ADMIN,
+        Permission.Action.CREATE,
+        Permission.Action.READ,
+        Permission.Action.WRITE);
+
+    SecureTestUtil.grantOnTable(TEST_UTIL, USER_RW.getShortName(),
+        TEST_TABLE, TEST_FAMILY, null,
+        Permission.Action.READ,
+        Permission.Action.WRITE);
+
+    // USER_CREATE is USER_RW plus CREATE permissions
+    SecureTestUtil.grantOnTable(TEST_UTIL, USER_CREATE.getShortName(),
+        TEST_TABLE, null, null,
+        Permission.Action.CREATE,
+        Permission.Action.READ,
+        Permission.Action.WRITE);
+
+    SecureTestUtil.grantOnTable(TEST_UTIL, USER_RO.getShortName(),
+        TEST_TABLE, TEST_FAMILY, null,
+        Permission.Action.READ);
+
+    SecureTestUtil.grantOnTable(TEST_UTIL, USER_ADMIN_CF.getShortName(),
+        TEST_TABLE, TEST_FAMILY,
+        null, Permission.Action.ADMIN, Permission.Action.CREATE);
+
+    SecureTestUtil.grantGlobal(TEST_UTIL, toGroupEntry(GROUP_ADMIN), Permission.Action.ADMIN);
+    SecureTestUtil.grantGlobal(TEST_UTIL, toGroupEntry(GROUP_CREATE), Permission.Action.CREATE);
+    SecureTestUtil.grantGlobal(TEST_UTIL, toGroupEntry(GROUP_READ), Permission.Action.READ);
+    SecureTestUtil.grantGlobal(TEST_UTIL, toGroupEntry(GROUP_WRITE), Permission.Action.WRITE);
+
+    assertEquals(5, AccessControlLists.getTablePermissions(conf, TEST_TABLE).size());
+    try {
+      assertEquals(5, AccessControlClient.getUserPermissions(systemUserConnection,
+          TEST_TABLE.toString()).size());
+    } catch (Throwable e) {
+      LOG.error("error during call of AccessControlClient.getUserPermissions. ", e);
+    }
+  }
+
+  private static void cleanUp() throws Exception {
+    // Clean the _acl_ table
+    try {
+      SecureTestUtil.deleteTable(TEST_UTIL, TEST_TABLE);
+    } catch (TableNotFoundException ex) {
+      // Test deleted the table, no problem
+      LOG.info("Test deleted table " + TEST_TABLE);
+    }
+    // Verify all table/namespace permissions are erased
+    assertEquals(0, AccessControlLists.getTablePermissions(conf, TEST_TABLE).size());
+    assertEquals(
+        0,
+        AccessControlLists.getNamespacePermissions(conf,
+            TEST_TABLE.getNamespaceAsString()).size());
+  }
+
+  @AfterClass
+  public static void tearDownAfterClass() throws Exception {
+    cleanUp();
+    TEST_UTIL.shutdownMiniCluster();
+    int total = TableAuthManager.getTotalRefCount();
+    assertTrue("Unexpected reference count: " + total, total == 0);
+  }
+
+  private static void configureRSGroupAdminEndpoint(Configuration conf) throws IOException {
+    String currentCoprocessors = conf.get(CoprocessorHost.MASTER_COPROCESSOR_CONF_KEY);
+    String coprocessors = RSGroupAdminEndpoint.class.getName();
+    if (currentCoprocessors != null) {
+      coprocessors += "," + currentCoprocessors;
+    }
+    conf.set(CoprocessorHost.MASTER_COPROCESSOR_CONF_KEY, coprocessors);
+    conf.set(HConstants.HBASE_MASTER_LOADBALANCER_CLASS,
+        RSGroupBasedLoadBalancer.class.getName());
+  }
+
+  @Test
+  public void testGetRSGroupInfo() throws Exception {
+    AccessTestAction action = new AccessTestAction() {
+      @Override
+      public Object run() throws Exception {
+        rsGroupAdminEndpoint.checkPermission("getRSGroupInfo");
+        return null;
+      }
+    };
+
+    SecureTestUtil.verifyAllowed(action, SUPERUSER, USER_ADMIN, USER_GROUP_ADMIN);
+    SecureTestUtil.verifyDenied(action, USER_CREATE, USER_OWNER, USER_RW, USER_RO,
+        USER_NONE, USER_GROUP_READ, USER_GROUP_WRITE, USER_GROUP_CREATE);
+  }
+
+  @Test
+  public void testGetRSGroupInfoOfTable() throws Exception {
+    AccessTestAction action = new AccessTestAction() {
+      @Override
+      public Object run() throws Exception {
+        rsGroupAdminEndpoint.checkPermission("getRSGroupInfoOfTable");
+        return null;
+      }
+    };
+
+    SecureTestUtil.verifyAllowed(action, SUPERUSER, USER_ADMIN, USER_GROUP_ADMIN);
+    SecureTestUtil.verifyDenied(action, USER_CREATE, USER_OWNER, USER_RW, USER_RO,
+        USER_NONE, USER_GROUP_READ, USER_GROUP_WRITE, USER_GROUP_CREATE);
+  }
+
+  @Test
+  public void testMoveServers() throws Exception {
+    AccessTestAction action = new AccessTestAction() {
+      @Override
+      public Object run() throws Exception {
+        rsGroupAdminEndpoint.checkPermission("moveServers");
+        return null;
+      }
+    };
+
+    SecureTestUtil.verifyAllowed(action, SUPERUSER, USER_ADMIN, USER_GROUP_ADMIN);
+    SecureTestUtil.verifyDenied(action, USER_CREATE, USER_OWNER, USER_RW, USER_RO,
+        USER_NONE, USER_GROUP_READ, USER_GROUP_WRITE, USER_GROUP_CREATE);
+  }
+
+  @Test
+  public void testMoveTables() throws Exception {
+    AccessTestAction action = new AccessTestAction() {
+      @Override
+      public Object run() throws Exception {
+        rsGroupAdminEndpoint.checkPermission("moveTables");
+        return null;
+      }
+    };
+
+    SecureTestUtil.verifyAllowed(action, SUPERUSER, USER_ADMIN, USER_GROUP_ADMIN);
+    SecureTestUtil.verifyDenied(action, USER_CREATE, USER_OWNER, USER_RW, USER_RO,
+        USER_NONE, USER_GROUP_READ, USER_GROUP_WRITE, USER_GROUP_CREATE);
+  }
+
+  @Test
+  public void testAddRSGroup() throws Exception {
+    AccessTestAction action = new AccessTestAction() {
+      @Override
+      public Object run() throws Exception {
+        rsGroupAdminEndpoint.checkPermission("addRSGroup");
+        return null;
+      }
+    };
+
+    SecureTestUtil.verifyAllowed(action, SUPERUSER, USER_ADMIN, USER_GROUP_ADMIN);
+    SecureTestUtil.verifyDenied(action, USER_CREATE, USER_OWNER, USER_RW, USER_RO,
+        USER_NONE, USER_GROUP_READ, USER_GROUP_WRITE, USER_GROUP_CREATE);
+  }
+
+  @Test
+  public void testRemoveRSGroup() throws Exception {
+    AccessTestAction action = new AccessTestAction() {
+      @Override
+      public Object run() throws Exception {
+        rsGroupAdminEndpoint.checkPermission("removeRSGroup");
+        return null;
+      }
+    };
+
+    SecureTestUtil.verifyAllowed(action, SUPERUSER, USER_ADMIN, USER_GROUP_ADMIN);
+    SecureTestUtil.verifyDenied(action, USER_CREATE, USER_OWNER, USER_RW, USER_RO,
+        USER_NONE, USER_GROUP_READ, USER_GROUP_WRITE, USER_GROUP_CREATE);
+  }
+
+  @Test
+  public void testBalanceRSGroup() throws Exception {
+    AccessTestAction action = new AccessTestAction() {
+      @Override
+      public Object run() throws Exception {
+        rsGroupAdminEndpoint.checkPermission("balanceRSGroup");
+        return null;
+      }
+    };
+
+    SecureTestUtil.verifyAllowed(action, SUPERUSER, USER_ADMIN, USER_GROUP_ADMIN);
+    SecureTestUtil.verifyDenied(action, USER_CREATE, USER_OWNER, USER_RW, USER_RO,
+        USER_NONE, USER_GROUP_READ, USER_GROUP_WRITE, USER_GROUP_CREATE);
+  }
+
+  @Test
+  public void testListRSGroup() throws Exception {
+    AccessTestAction action = new AccessTestAction() {
+      @Override
+      public Object run() throws Exception {
+        rsGroupAdminEndpoint.checkPermission("listRSGroup");
+        return null;
+      }
+    };
+
+    SecureTestUtil.verifyAllowed(action, SUPERUSER, USER_ADMIN, USER_GROUP_ADMIN);
+    SecureTestUtil.verifyDenied(action, USER_CREATE, USER_OWNER, USER_RW, USER_RO,
+        USER_NONE, USER_GROUP_READ, USER_GROUP_WRITE, USER_GROUP_CREATE);
+  }
+
+  @Test
+  public void testGetRSGroupInfoOfServer() throws Exception {
+    AccessTestAction action = new AccessTestAction() {
+      @Override
+      public Object run() throws Exception {
+        rsGroupAdminEndpoint.checkPermission("getRSGroupInfoOfServer");
+        return null;
+      }
+    };
+
+    SecureTestUtil.verifyAllowed(action, SUPERUSER, USER_ADMIN, USER_GROUP_ADMIN);
+    SecureTestUtil.verifyDenied(action, USER_CREATE, USER_OWNER, USER_RW, USER_RO,
+        USER_NONE, USER_GROUP_READ, USER_GROUP_WRITE, USER_GROUP_CREATE);
+  }
+
+  @Test
+  public void testMoveServersAndTables() throws Exception {
+    AccessTestAction action = new AccessTestAction() {
+      @Override
+      public Object run() throws Exception {
+        rsGroupAdminEndpoint.checkPermission("moveServersAndTables");
+        return null;
+      }
+    };
+
+    SecureTestUtil.verifyAllowed(action, SUPERUSER, USER_ADMIN, USER_GROUP_ADMIN);
+    SecureTestUtil.verifyDenied(action, USER_CREATE, USER_OWNER, USER_RW, USER_RO,
+        USER_NONE, USER_GROUP_READ, USER_GROUP_WRITE, USER_GROUP_CREATE);
+  }
+}

hbase-server/src/main/java/org/apache/hadoop/hbase/master/MasterRpcServices.java

@@ -83,6 +83,7 @@ import org.apache.hadoop.hbase.replication.ReplicationException;
 import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
 import org.apache.hadoop.hbase.replication.ReplicationPeerDescription;
 import org.apache.hadoop.hbase.security.User;
+import org.apache.hadoop.hbase.security.access.AccessChecker;
 import org.apache.hadoop.hbase.security.access.AccessController;
 import org.apache.hadoop.hbase.security.visibility.VisibilityController;
 import org.apache.hadoop.hbase.snapshot.ClientSnapshotDescriptionUtils;
@@ -1810,7 +1811,7 @@ public class MasterRpcServices extends RSRpcServices
       // A coprocessor that implements AccessControlService can provide AUTHORIZATION and
       // CELL_AUTHORIZATION
       if (master.cpHost != null && hasAccessControlServiceCoprocessor(master.cpHost)) {
-        if (AccessController.isAuthorizationSupported(master.getConfiguration())) {
+        if (AccessChecker.isAuthorizationSupported(master.getConfiguration())) {
           capabilities.add(SecurityCapabilitiesResponse.Capability.AUTHORIZATION);
         }
         if (AccessController.isCellAuthorizationSupported(master.getConfiguration())) {

hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/AccessChecker.java

@@ -0,0 +1,332 @@
+/*
+ * 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.security.access;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.util.Collection;
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.DoNotRetryIOException;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.hadoop.hbase.ipc.RpcServer;
+import org.apache.hadoop.hbase.security.AccessDeniedException;
+import org.apache.hadoop.hbase.security.User;
+import org.apache.hadoop.hbase.security.access.Permission.Action;
+import org.apache.hadoop.hbase.zookeeper.ZKWatcher;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@InterfaceAudience.Private
+public final class AccessChecker {
+  private static final Logger AUDITLOG =
+      LoggerFactory.getLogger("SecurityLogger." + AccessChecker.class.getName());
+  private TableAuthManager authManager;
+  /**
+   * if we are active, usually false, only true if "hbase.security.authorization"
+   * has been set to true in site configuration.see HBASE-19483.
+   */
+  private boolean authorizationEnabled;
+
+  public static boolean isAuthorizationSupported(Configuration conf) {
+    return conf.getBoolean(User.HBASE_SECURITY_AUTHORIZATION_CONF_KEY, false);
+  }
+
+  /**
+   * Constructor with existing configuration
+   *
+   * @param conf Existing configuration to use
+   * @param zkw reference to the {@link ZKWatcher}
+   */
+  public AccessChecker(final Configuration conf, final ZKWatcher zkw)
+      throws RuntimeException {
+    // If zk is null or IOException while obtaining auth manager,
+    // throw RuntimeException so that the coprocessor is unloaded.
+    if (zkw != null) {
+      try {
+        this.authManager = TableAuthManager.getOrCreate(zkw, conf);
+      } catch (IOException ioe) {
+        throw new RuntimeException("Error obtaining AccessChecker", ioe);
+      }
+    } else {
+      throw new NullPointerException("Error obtaining AccessChecker, zk found null.");
+    }
+    authorizationEnabled = isAuthorizationSupported(conf);
+  }
+
+  public TableAuthManager getAuthManager() {
+    return authManager;
+  }
+
+  /**
+   * Authorizes that the current user has any of the given permissions to access the table.
+   *
+   * @param tableName   Table requested
+   * @param permissions Actions being requested
+   * @throws IOException           if obtaining the current user fails
+   * @throws AccessDeniedException if user has no authorization
+   */
+  public void requireAccess(User user, String request, TableName tableName,
+      Action... permissions) throws IOException {
+    AuthResult result = null;
+
+    for (Action permission : permissions) {
+      if (authManager.hasAccess(user, tableName, permission)) {
+        result = AuthResult.allow(request, "Table permission granted",
+            user, permission, tableName, null, null);
+        break;
+      } else {
+        // rest of the world
+        result = AuthResult.deny(request, "Insufficient permissions",
+            user, permission, tableName, null, null);
+      }
+    }
+    logResult(result);
+    if (authorizationEnabled && !result.isAllowed()) {
+      throw new AccessDeniedException("Insufficient permissions " + result.toContextString());
+    }
+  }
+
+  /**
+   * Authorizes that the current user has global privileges for the given action.
+   *
+   * @param perm The action being requested
+   * @throws IOException           if obtaining the current user fails
+   * @throws AccessDeniedException if authorization is denied
+   */
+  public void requirePermission(User user, String request, Action perm)
+      throws IOException {
+    requireGlobalPermission(user, request, perm, null, null);
+  }
+
+  /**
+   * Checks that the user has the given global permission. The generated
+   * audit log message will contain context information for the operation
+   * being authorized, based on the given parameters.
+   *
+   * @param perm      Action being requested
+   * @param tableName Affected table name.
+   * @param familyMap Affected column families.
+   */
+  public void requireGlobalPermission(User user, String request,
+      Action perm, TableName tableName,
+      Map<byte[], ? extends Collection<byte[]>> familyMap)throws IOException {
+    AuthResult result;
+    if (authManager.authorize(user, perm)) {
+      result = AuthResult.allow(request, "Global check allowed",
+          user, perm, tableName, familyMap);
+      result.getParams().setTableName(tableName).setFamilies(familyMap);
+      logResult(result);
+    } else {
+      result = AuthResult.deny(request, "Global check failed",
+          user, perm, tableName, familyMap);
+      result.getParams().setTableName(tableName).setFamilies(familyMap);
+      logResult(result);
+      if (authorizationEnabled) {
+        throw new AccessDeniedException(
+            "Insufficient permissions for user '" + (user != null ? user.getShortName() : "null")
+                + "' (global, action=" + perm.toString() + ")");
+      }
+    }
+  }
+
+  /**
+   * Checks that the user has the given global permission. The generated
+   * audit log message will contain context information for the operation
+   * being authorized, based on the given parameters.
+   *
+   * @param perm      Action being requested
+   * @param namespace The given namespace
+   */
+  public void requireGlobalPermission(User user, String request, Action perm,
+      String namespace) throws IOException {
+    AuthResult authResult;
+    if (authManager.authorize(user, perm)) {
+      authResult = AuthResult.allow(request, "Global check allowed",
+          user, perm, null);
+      authResult.getParams().setNamespace(namespace);
+      logResult(authResult);
+    } else {
+      authResult = AuthResult.deny(request, "Global check failed",
+          user, perm, null);
+      authResult.getParams().setNamespace(namespace);
+      logResult(authResult);
+      if (authorizationEnabled) {
+        throw new AccessDeniedException(
+            "Insufficient permissions for user '" + (user != null ? user.getShortName() : "null")
+                + "' (global, action=" + perm.toString() + ")");
+      }
+    }
+  }
+
+  /**
+   * Checks that the user has the given global or namespace permission.
+   *
+   * @param namespace  The given namespace
+   * @param permissions Actions being requested
+   */
+  public void requireNamespacePermission(User user, String request, String namespace,
+      Action... permissions) throws IOException {
+    AuthResult result = null;
+
+    for (Action permission : permissions) {
+      if (authManager.authorize(user, namespace, permission)) {
+        result =
+            AuthResult.allow(request, "Namespace permission granted",
+                user, permission, namespace);
+        break;
+      } else {
+        // rest of the world
+        result = AuthResult.deny(request, "Insufficient permissions",
+            user, permission, namespace);
+      }
+    }
+    logResult(result);
+    if (authorizationEnabled && !result.isAllowed()) {
+      throw new AccessDeniedException("Insufficient permissions " + result.toContextString());
+    }
+  }
+
+  /**
+   * Checks that the user has the given global or namespace permission.
+   *
+   * @param namespace  The given namespace
+   * @param permissions Actions being requested
+   */
+  public void requireNamespacePermission(User user, String request, String namespace,
+      TableName tableName, Map<byte[], ? extends Collection<byte[]>> familyMap,
+      Action... permissions) throws IOException {
+    AuthResult result = null;
+
+    for (Action permission : permissions) {
+      if (authManager.authorize(user, namespace, permission)) {
+        result =
+            AuthResult.allow(request, "Namespace permission granted",
+                user, permission, namespace);
+        result.getParams().setTableName(tableName).setFamilies(familyMap);
+        break;
+      } else {
+        // rest of the world
+        result = AuthResult.deny(request, "Insufficient permissions",
+            user, permission, namespace);
+        result.getParams().setTableName(tableName).setFamilies(familyMap);
+      }
+    }
+    logResult(result);
+    if (authorizationEnabled && !result.isAllowed()) {
+      throw new AccessDeniedException("Insufficient permissions " + result.toContextString());
+    }
+  }
+
+  /**
+   * Authorizes that the current user has any of the given permissions for the
+   * given table, column family and column qualifier.
+   *
+   * @param tableName Table requested
+   * @param family    Column family requested
+   * @param qualifier Column qualifier requested
+   * @throws IOException           if obtaining the current user fails
+   * @throws AccessDeniedException if user has no authorization
+   */
+  public void requirePermission(User user, String request, TableName tableName, byte[] family,
+      byte[] qualifier, Action... permissions) throws IOException {
+    AuthResult result = null;
+
+    for (Action permission : permissions) {
+      if (authManager.authorize(user, tableName, family, qualifier, permission)) {
+        result = AuthResult.allow(request, "Table permission granted",
+            user, permission, tableName, family,
+                qualifier);
+        break;
+      } else {
+        // rest of the world
+        result = AuthResult.deny(request, "Insufficient permissions",
+                user, permission, tableName, family,
+                qualifier);
+      }
+    }
+    logResult(result);
+    if (authorizationEnabled && !result.isAllowed()) {
+      throw new AccessDeniedException("Insufficient permissions " + result.toContextString());
+    }
+  }
+
+  /**
+   * Authorizes that the current user has any of the given permissions for the
+   * given table, column family and column qualifier.
+   *
+   * @param tableName Table requested
+   * @param family    Column family param
+   * @param qualifier Column qualifier param
+   * @throws IOException           if obtaining the current user fails
+   * @throws AccessDeniedException if user has no authorization
+   */
+  public void requireTablePermission(User user, String request,
+      TableName tableName,byte[] family, byte[] qualifier,
+      Action... permissions) throws IOException {
+    AuthResult result = null;
+
+    for (Action permission : permissions) {
+      if (authManager.authorize(user, tableName, null, null, permission)) {
+        result = AuthResult.allow(request, "Table permission granted",
+            user, permission, tableName, null, null);
+        result.getParams().setFamily(family).setQualifier(qualifier);
+        break;
+      } else {
+        // rest of the world
+        result = AuthResult.deny(request, "Insufficient permissions",
+                user, permission, tableName, family, qualifier);
+        result.getParams().setFamily(family).setQualifier(qualifier);
+      }
+    }
+    logResult(result);
+    if (authorizationEnabled && !result.isAllowed()) {
+      throw new AccessDeniedException("Insufficient permissions " + result.toContextString());
+    }
+  }
+
+  public void checkLockPermissions(User user, String namespace,
+      TableName tableName, RegionInfo[] regionInfos, String reason)
+      throws IOException {
+    if (namespace != null && !namespace.isEmpty()) {
+      requireNamespacePermission(user, reason, namespace, Action.ADMIN, Action.CREATE);
+    } else if (tableName != null || (regionInfos != null && regionInfos.length > 0)) {
+      // So, either a table or regions op. If latter, check perms ons table.
+      TableName tn = tableName != null? tableName: regionInfos[0].getTable();
+      requireTablePermission(user, reason, tn, null, null,
+          Action.ADMIN, Action.CREATE);
+    } else {
+      throw new DoNotRetryIOException("Invalid lock level when requesting permissions.");
+    }
+  }
+
+  public static void logResult(AuthResult result) {
+    if (AUDITLOG.isTraceEnabled()) {
+      AUDITLOG.trace("Access " + (result.isAllowed() ? "allowed" : "denied") + " for user " + (
+          result.getUser() != null ?
+              result.getUser().getShortName() :
+              "UNKNOWN") + "; reason: " + result.getReason() + "; remote address: "
+          + RpcServer.getRemoteAddress().map(InetAddress::toString).orElse("")
+          + "; request: " + result.getRequest() + "; context: " + result.toContextString());
+    }
+  }
+}

hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/AccessControlLists.java

@@ -490,7 +490,7 @@ public class AccessControlLists {
     return getPermissions(conf, tableName != null ? tableName.getName() : null, null);
   }
 
-  static ListMultimap<String, TablePermission> getNamespacePermissions(Configuration conf,
+  public static ListMultimap<String, TablePermission> getNamespacePermissions(Configuration conf,
       String namespace) throws IOException {
     return getPermissions(conf, Bytes.toBytes(toNamespaceEntry(namespace)), null);
   }

hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/TableAuthManager.java

@@ -754,7 +754,7 @@ public class TableAuthManager implements Closeable {
   }
 
   @VisibleForTesting
-  static int getTotalRefCount() {
+  public static int getTotalRefCount() {
     int total = 0;
     for (int count : refCount.values()) {
       total += count;

hbase-server/src/main/java/org/apache/hadoop/hbase/security/visibility/VisibilityController.java

@@ -105,6 +105,7 @@ import org.apache.hadoop.hbase.regionserver.querymatcher.DeleteTracker;
 import org.apache.hadoop.hbase.security.AccessDeniedException;
 import org.apache.hadoop.hbase.security.Superusers;
 import org.apache.hadoop.hbase.security.User;
+import org.apache.hadoop.hbase.security.access.AccessChecker;
 import org.apache.hadoop.hbase.security.access.AccessController;
 import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
 import org.apache.hbase.thirdparty.com.google.common.collect.MapMaker;
@@ -141,8 +142,8 @@ public class VisibilityController implements MasterCoprocessor, RegionCoprocesso
 
   private VisibilityLabelService visibilityLabelService;
 
-  /** if we are active, usually true, only not true if "hbase.security.authorization"
+  /** if we are active, usually false, only true if "hbase.security.authorization"
-    has been set to false in site configuration */
+    has been set to true in site configuration */
   boolean authorizationEnabled;
 
   // Add to this list if there are any reserved tag types
@@ -153,19 +154,15 @@ public class VisibilityController implements MasterCoprocessor, RegionCoprocesso
     RESERVED_VIS_TAG_TYPES.add(TagType.STRING_VIS_TAG_TYPE);
   }
 
-  public static boolean isAuthorizationSupported(Configuration conf) {
-    return conf.getBoolean(User.HBASE_SECURITY_AUTHORIZATION_CONF_KEY, true);
-  }
-
   public static boolean isCellAuthorizationSupported(Configuration conf) {
-    return isAuthorizationSupported(conf);
+    return AccessChecker.isAuthorizationSupported(conf);
   }
 
   @Override
   public void start(CoprocessorEnvironment env) throws IOException {
     this.conf = env.getConfiguration();
 
-    authorizationEnabled = isAuthorizationSupported(conf);
+    authorizationEnabled = AccessChecker.isAuthorizationSupported(conf);
     if (!authorizationEnabled) {
       LOG.warn("The VisibilityController has been loaded with authorization checks disabled.");
     }

hbase-server/src/test/java/org/apache/hadoop/hbase/security/access/SecureTestUtil.java

@@ -106,6 +106,7 @@ public class SecureTestUtil {
     conf.set(CoprocessorHost.REGIONSERVER_COPROCESSOR_CONF_KEY, AccessController.class.getName());
     // Need HFile V3 for tags for security features
     conf.setInt(HFile.FORMAT_VERSION_KEY, 3);
+    conf.set(User.HBASE_SECURITY_AUTHORIZATION_CONF_KEY, "true");
     configureSuperuser(conf);
   }
 
@@ -129,6 +130,11 @@ public class SecureTestUtil {
     if (conf.getInt(HFile.FORMAT_VERSION_KEY, 2) < HFile.MIN_FORMAT_VERSION_WITH_TAGS) {
       throw new RuntimeException("Post 0.96 security features require HFile version >= 3");
     }
+
+    if (!conf.getBoolean(User.HBASE_SECURITY_AUTHORIZATION_CONF_KEY, false)) {
+      throw new RuntimeException("Post 2.0.0 security features require set "
+          + User.HBASE_SECURITY_AUTHORIZATION_CONF_KEY + " to true");
+    }
   }
 
   public static void checkTablePerms(Configuration conf, TableName table, byte[] family, byte[] column,

hbase-server/src/test/java/org/apache/hadoop/hbase/security/access/TestAccessController.java

@@ -2838,81 +2838,6 @@ public class TestAccessController extends SecureTestUtil {
       USER_GROUP_READ, USER_GROUP_ADMIN, USER_GROUP_CREATE);
   }
 
-  @Test
-  public void testMoveServers() throws Exception {
-    AccessTestAction action1 = new AccessTestAction() {
-      @Override
-      public Object run() throws Exception {
-        ACCESS_CONTROLLER.preMoveServers(ObserverContextImpl.createAndPrepare(CP_ENV),
-            null, null);
-        return null;
-      }
-    };
-
-    verifyAllowed(action1, SUPERUSER, USER_ADMIN);
-    verifyDenied(action1, USER_CREATE, USER_RW, USER_RO, USER_NONE, USER_OWNER);
-  }
-
-  @Test
-  public void testMoveTables() throws Exception {
-    AccessTestAction action1 = new AccessTestAction() {
-      @Override
-      public Object run() throws Exception {
-        ACCESS_CONTROLLER.preMoveTables(ObserverContextImpl.createAndPrepare(CP_ENV),
-            null, null);
-        return null;
-      }
-    };
-
-    verifyAllowed(action1, SUPERUSER, USER_ADMIN);
-    verifyDenied(action1, USER_CREATE, USER_RW, USER_RO, USER_NONE, USER_OWNER);
-  }
-
-  @Test
-  public void testAddGroup() throws Exception {
-    AccessTestAction action1 = new AccessTestAction() {
-      @Override
-      public Object run() throws Exception {
-        ACCESS_CONTROLLER.preAddRSGroup(ObserverContextImpl.createAndPrepare(CP_ENV),
-            null);
-        return null;
-      }
-    };
-
-    verifyAllowed(action1, SUPERUSER, USER_ADMIN);
-    verifyDenied(action1, USER_CREATE, USER_RW, USER_RO, USER_NONE, USER_OWNER);
-  }
-
-  @Test
-  public void testRemoveGroup() throws Exception {
-    AccessTestAction action1 = new AccessTestAction() {
-      @Override
-      public Object run() throws Exception {
-        ACCESS_CONTROLLER.preRemoveRSGroup(ObserverContextImpl.createAndPrepare(CP_ENV),
-            null);
-        return null;
-      }
-    };
-
-    verifyAllowed(action1, SUPERUSER, USER_ADMIN);
-    verifyDenied(action1, USER_CREATE, USER_RW, USER_RO, USER_NONE, USER_OWNER);
-  }
-
-  @Test
-  public void testBalanceGroup() throws Exception {
-    AccessTestAction action1 = new AccessTestAction() {
-      @Override
-      public Object run() throws Exception {
-        ACCESS_CONTROLLER.preBalanceRSGroup(ObserverContextImpl.createAndPrepare(CP_ENV),
-            null);
-        return null;
-      }
-    };
-
-    verifyAllowed(action1, SUPERUSER, USER_ADMIN);
-    verifyDenied(action1, USER_CREATE, USER_RW, USER_RO, USER_NONE, USER_OWNER);
-  }
-
   @Test
   public void testAddReplicationPeer() throws Exception {
     AccessTestAction action = new AccessTestAction() {

hbase-server/src/test/java/org/apache/hadoop/hbase/security/visibility/VisibilityTestUtil.java

@@ -14,7 +14,7 @@ import java.io.IOException;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
-
+import org.apache.hadoop.hbase.security.User;
 
 /**
  * Utility methods for testing visibility labels.
@@ -23,6 +23,7 @@ public class VisibilityTestUtil {
 
   public static void enableVisiblityLabels(Configuration conf) throws IOException {
     conf.setInt("hfile.format.version", 3);
+    conf.setBoolean(User.HBASE_SECURITY_AUTHORIZATION_CONF_KEY, true);
     appendCoprocessor(conf, CoprocessorHost.MASTER_COPROCESSOR_CONF_KEY,
       VisibilityController.class.getName());
     appendCoprocessor(conf, CoprocessorHost.REGION_COPROCESSOR_CONF_KEY,

hbase-thrift/src/test/java/org/apache/hadoop/hbase/thrift2/TestThriftHBaseServiceHandlerWithLabels.java

@@ -47,7 +47,7 @@ import org.apache.hadoop.hbase.security.visibility.ScanLabelGenerator;
 import org.apache.hadoop.hbase.security.visibility.SimpleScanLabelGenerator;
 import org.apache.hadoop.hbase.security.visibility.VisibilityClient;
 import org.apache.hadoop.hbase.security.visibility.VisibilityConstants;
-import org.apache.hadoop.hbase.security.visibility.VisibilityController;
+import org.apache.hadoop.hbase.security.visibility.VisibilityTestUtil;
 import org.apache.hadoop.hbase.security.visibility.VisibilityUtils;
 import org.apache.hadoop.hbase.testclassification.ClientTests;
 import org.apache.hadoop.hbase.testclassification.MediumTests;
@@ -131,11 +131,7 @@ public class TestThriftHBaseServiceHandlerWithLabels {
     conf.setClass(VisibilityUtils.VISIBILITY_LABEL_GENERATOR_CLASS,
         SimpleScanLabelGenerator.class, ScanLabelGenerator.class);
     conf.set("hbase.superuser", SUPERUSER.getShortName());
-    conf.set("hbase.coprocessor.master.classes",
+    VisibilityTestUtil.enableVisiblityLabels(conf);
-        VisibilityController.class.getName());
-    conf.set("hbase.coprocessor.region.classes",
-        VisibilityController.class.getName());
-    conf.setInt("hfile.format.version", 3);
     UTIL.startMiniCluster(1);
     // Wait for the labels table to become available
     UTIL.waitTableEnabled(VisibilityConstants.LABELS_TABLE_NAME.getName(), 50000);

src/main/asciidoc/_chapters/appendix_acl_matrix.adoc

@@ -164,6 +164,17 @@ In case the table goes out of date, the unit tests which check for accuracy of p
 |              | mergeRegions | superuser\|global(A)
 |              | rollWALWriterRequest | superuser\|global(A)
 |              | replicateLogEntries | superuser\|global(W)
+|RSGroup |addRSGroup           |superuser\|global(A)
+|        |balanceRSGroup       |superuser\|global(A)
+|        |getRSGroupInfo       |superuser\|global(A)
+|        |getRSGroupInfoOfTable|superuser\|global(A)
+|        |getRSGroupOfServer   |superuser\|global(A)
+|        |listRSGroups         |superuser\|global(A)
+|        |moveServers          |superuser\|global(A)
+|        |moveServersAndTables |superuser\|global(A)
+|        |moveTables           |superuser\|global(A)
+|        |removeRSGroup        |superuser\|global(A)
+|        |removeServers        |superuser\|global(A)
 |===
 
 :numbered:

src/main/asciidoc/_chapters/ops_mgt.adoc

@@ -2703,6 +2703,18 @@ Viewing the Master log will give you insight on rsgroup operation.
 
 If it appears stuck, restart the Master process.
 
+=== ACL
+To enable ACL, add the following to your hbase-site.xml and restart your Master:
+
+[source,xml]
+----
+<property>
+  <name>hbase.security.authorization</name>
+  <value>true</value>
+<property>
+----
+
+
 
 [[normalizer]]
 == Region Normalizer

src/main/asciidoc/_chapters/security.adoc

@@ -806,6 +806,10 @@ For an example of using both together, see <<security.example.config>>.
 +
 [source,xml]
 ----
+<property>
+  <name>hbase.security.authorization</name>
+  <value>true</value>
+</property>
 <property>
   <name>hbase.coprocessor.region.classes</name>
   <value>org.apache.hadoop.hbase.security.access.AccessController, org.apache.hadoop.hbase.security.token.TokenProvider</value>
@@ -1186,6 +1190,10 @@ NOTE: Visibility labels are not currently applied for superusers.
 +
 [source,xml]
 ----
+<property>
+  <name>hbase.security.authorization</name>
+  <value>true</value>
+</property>
 <property>
   <name>hbase.coprocessor.region.classes</name>
   <value>org.apache.hadoop.hbase.security.visibility.VisibilityController</value>
@@ -1453,6 +1461,10 @@ As mentioned in the above sections, the interface `VisibilityLabelService` could
 
 [source,xml]
 ----
+<property>
+  <name>hbase.security.authorization</name>
+  <value>true</value>
+</property>
 <property>
   <name>hbase.coprocessor.regionserver.classes</name>
   <value>org.apache.hadoop.hbase.security.visibility.VisibilityController$VisibilityReplication</value>
@@ -1671,6 +1683,10 @@ To enable secure bulk load, add the following properties to _hbase-site.xml_.
 
 [source,xml]
 ----
+<property>
+  <name>hbase.security.authorization</name>
+  <value>true</value>
+</property>
 <property>
   <name>hbase.bulkload.staging.dir</name>
   <value>/tmp/hbase-staging</value>
@@ -1682,6 +1698,22 @@ To enable secure bulk load, add the following properties to _hbase-site.xml_.
 </property>
 ----
 
+[[hbase.secure.enable]]
+=== Secure Enable
+After hbase-2.x, the default 'hbase.security.authorization' changed.
+Before hbase-2.x, it defaulted to true, in later HBase versions, the
+default became false.
+So to enable hbase authorization, the following propertie must be configured in _hbase-site.xml_.
+See link:https://issues.apache.org/jira/browse/HBASE-19483[HBASE-19483];
+
+[source,xml]
+----
+<property>
+  <name>hbase.security.authorization</name>
+  <value>true</value>
+</property>
+----
+
 [[security.example.config]]
 == Security Configuration Example
 
@@ -1703,6 +1735,10 @@ All options have been discussed separately in the sections above.
   <value>hbase, admin</value>
 </property>
 <!-- Coprocessors for ACLs and Visibility Tags -->
+<property>
+  <name>hbase.security.authorization</name>
+  <value>true</value>
+</property>
 <property>
   <name>hbase.coprocessor.region.classes</name>
   <value>org.apache.hadoop.hbase.security.access.AccessController,