HBASE-18437 Revoke access permissions of a user from a table does not work as expected

Signed-off-by: Andrew Purtell <apurtell@apache.org>

hbase-client/src/main/java/org/apache/hadoop/hbase/security/access/Permission.java

@@ -110,6 +110,12 @@ public class Permission extends VersionedWritable {
     return false;
   }
 
+  public void setActions(Action[] assigned) {
+    if (assigned != null && assigned.length > 0) {
+      actions = Arrays.copyOf(assigned, assigned.length);
+    }
+  }
+
   @Override
   public boolean equals(Object obj) {
     if (!(obj instanceof Permission)) {

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

@@ -241,13 +241,40 @@ public class AccessControlLists {
    */
   static void removeUserPermission(Configuration conf, UserPermission userPerm, Table t)
       throws IOException {
-    Delete d = new Delete(userPermissionRowKey(userPerm));
+    if (null == userPerm.getActions()) {
-    byte[] key = userPermissionKey(userPerm);
+      removePermissionRecord(conf, userPerm, t);
-
+    } else {
-    if (LOG.isDebugEnabled()) {
+      // Get all the global user permissions from the acl table
-      LOG.debug("Removing permission "+ userPerm.toString());
+      List<UserPermission> permsList = getUserPermissions(conf, userPermissionRowKey(userPerm));
+      List<Permission.Action> remainingActions = new ArrayList<>();
+      List<Permission.Action> dropActions = Arrays.asList(userPerm.getActions());
+      for (UserPermission perm : permsList) {
+        // Find the user and remove only the requested permissions
+        if (Bytes.toString(perm.getUser()).equals(Bytes.toString(userPerm.getUser()))) {
+          for (Permission.Action oldAction : perm.getActions()) {
+            if (!dropActions.contains(oldAction)) {
+              remainingActions.add(oldAction);
+            }
+          }
+          if (!remainingActions.isEmpty()) {
+            perm.setActions(remainingActions.toArray(new Permission.Action[remainingActions.size()]));
+            addUserPermission(conf, perm, t);
+          } else {
+            removePermissionRecord(conf, userPerm, t);
+          }
+          break;
+        }
+      }
     }
-    d.addColumns(ACL_LIST_FAMILY, key);
+    if (LOG.isDebugEnabled()) {
+      LOG.debug("Removed permission "+ userPerm.toString());
+    }
+  }
+
+  private static void removePermissionRecord(Configuration conf, UserPermission userPerm, Table t)
+      throws IOException {
+    Delete d = new Delete(userPermissionRowKey(userPerm));
+    d.addColumns(ACL_LIST_FAMILY, userPermissionKey(userPerm));
     try {
       t.delete(d);
     } finally {

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

@@ -18,6 +18,7 @@
 package org.apache.hadoop.hbase.security.access;
 
 import static org.apache.hadoop.hbase.AuthUtil.toGroupEntry;
+import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
@@ -2380,28 +2381,30 @@ public class TestAccessController extends SecureTestUtil {
     // Grant table READ permissions to testGlobalGrantRevoke.
     String userName = testGlobalGrantRevoke.getShortName();
     try {
-      grantGlobalUsingAccessControlClient(TEST_UTIL, systemUserConnection,
+      grantGlobalUsingAccessControlClient(TEST_UTIL, systemUserConnection, userName,
-          userName, Permission.Action.READ);
+        Permission.Action.READ);
     } catch (Throwable e) {
       LOG.error("error during call of AccessControlClient.grant. ", e);
     }
     try {
       // Now testGlobalGrantRevoke should be able to read also
       verifyAllowed(getAction, testGlobalGrantRevoke);
-
+    } catch (Exception e) {
-      // Revoke table READ permission to testGlobalGrantRevoke.
-      try {
-        revokeGlobalUsingAccessControlClient(TEST_UTIL, systemUserConnection,
-          userName, Permission.Action.READ);
-      } catch (Throwable e) {
-        LOG.error("error during call of AccessControlClient.revoke ", e);
-      }
-
-      // Now testGlobalGrantRevoke shouldn't be able read
-      verifyDenied(getAction, testGlobalGrantRevoke);
-    } finally {
       revokeGlobal(TEST_UTIL, userName, Permission.Action.READ);
+      throw e;
     }
+
+    // Revoke table READ permission to testGlobalGrantRevoke.
+    try {
+      revokeGlobalUsingAccessControlClient(TEST_UTIL, systemUserConnection, userName,
+        Permission.Action.READ);
+    } catch (Throwable e) {
+      LOG.error("error during call of AccessControlClient.revoke ", e);
+    }
+
+    // Now testGlobalGrantRevoke shouldn't be able read
+    verifyDenied(getAction, testGlobalGrantRevoke);
+
   }
 
   @Test(timeout = 180000)
@@ -2547,28 +2550,29 @@ public class TestAccessController extends SecureTestUtil {
     String namespace = TEST_TABLE.getNamespaceAsString();
     // Grant namespace READ to testNS, this should supersede any table permissions
     try {
-      grantOnNamespaceUsingAccessControlClient(TEST_UTIL, systemUserConnection, userName,
+      grantOnNamespaceUsingAccessControlClient(TEST_UTIL, systemUserConnection, userName, namespace,
-        namespace, Permission.Action.READ);
+        Permission.Action.READ);
     } catch (Throwable e) {
       LOG.error("error during call of AccessControlClient.grant. ", e);
     }
     try {
       // Now testNS should be able to read also
       verifyAllowed(getAction, testNS);
-
+    } catch (Exception e) {
-      // Revoke namespace READ to testNS, this should supersede any table permissions
-      try {
-        revokeFromNamespaceUsingAccessControlClient(TEST_UTIL, systemUserConnection, userName,
-          namespace, Permission.Action.READ);
-      } catch (Throwable e) {
-        LOG.error("error during call of AccessControlClient.revoke ", e);
-      }
-
-      // Now testNS shouldn't be able read
-      verifyDenied(getAction, testNS);
-    } finally {
       revokeFromNamespace(TEST_UTIL, userName, namespace, Permission.Action.READ);
+      throw e;
     }
+
+    // Revoke namespace READ to testNS, this should supersede any table permissions
+    try {
+      revokeFromNamespaceUsingAccessControlClient(TEST_UTIL, systemUserConnection, userName,
+        namespace, Permission.Action.READ);
+    } catch (Throwable e) {
+      LOG.error("error during call of AccessControlClient.revoke ", e);
+    }
+
+    // Now testNS shouldn't be able read
+    verifyDenied(getAction, testNS);
   }
 
 
@@ -3175,4 +3179,40 @@ public class TestAccessController extends SecureTestUtil {
     verifyAllowed(regionLockHeartbeatAction, SUPERUSER, USER_ADMIN, namespaceUser, tableACUser);
     verifyDenied(regionLockHeartbeatAction, globalRWXUser, tableRWXUser);
   }
+
+  @Test
+  public void testAccessControlRevokeOnlyFewPermission() throws Throwable {
+    TableName tname = TableName.valueOf("revoke");
+    try {
+      TEST_UTIL.createTable(tname, TEST_FAMILY);
+      User testUserPerms = User.createUserForTesting(conf, "revokePerms", new String[0]);
+      Permission.Action[] actions = { Action.READ, Action.WRITE };
+      AccessControlClient.grant(TEST_UTIL.getConnection(), tname, testUserPerms.getShortName(),
+        null, null, actions);
+
+      List<UserPermission> userPermissions = AccessControlClient
+          .getUserPermissions(TEST_UTIL.getConnection(), tname.getNameAsString());
+      assertEquals(2, userPermissions.size());
+
+      AccessControlClient.revoke(TEST_UTIL.getConnection(), tname, testUserPerms.getShortName(),
+        null, null, Action.WRITE);
+
+      userPermissions = AccessControlClient.getUserPermissions(TEST_UTIL.getConnection(),
+        tname.getNameAsString());
+      assertEquals(2, userPermissions.size());
+
+      Permission.Action[] expectedAction = { Action.READ };
+      boolean userFound = false;
+      for (UserPermission p : userPermissions) {
+        if (testUserPerms.getShortName().equals(Bytes.toString(p.getUser()))) {
+          assertArrayEquals(expectedAction, p.getActions());
+          userFound = true;
+          break;
+        }
+      }
+      assertTrue(userFound);
+    } finally {
+      TEST_UTIL.deleteTable(tname);
+    }
+  }
 }