RANGER-5750:Handle authorization of hbase:acl table in RangerAuthorizationCoprocessor in HBase Ranger plugin

hbase-agent/src/main/java/org/apache/ranger/authorization/hbase/RangerAuthorizationCoprocessor.java

@@ -852,17 +852,34 @@ protected byte[] getTableName(RegionCoprocessorEnvironment e) {
         return tableName;
     }
 
+    private boolean isSystemOrSuperUser(User activeUser) {
+        if (activeUser == null) {
+            return false;
+        }
+        if (userUtils.isSuperUser(activeUser)) {
+            return true;
+        }
+        try {
+            User currentUser = User.getCurrent();
+            if (currentUser != null) {
+                return Objects.equals(currentUser.getShortName(), activeUser.getShortName());
+            }
+        } catch (IOException e) {
+            LOG.warn("Unable to obtain the current user", e);
+        }
+        return false;
+    }
+
     protected void requireSystemOrSuperUser(ObserverContext<?> ctx) throws IOException {
         User user = User.getCurrent();
 
         if (user == null) {
             throw new IOException("Unable to obtain the current user, authorization checks for internal operations will not work correctly!");
         }
 
-        String systemUser = user.getShortName();
-        User   activeUser = getActiveUser(ctx);
+        User activeUser = getActiveUser(ctx);
 
-        if (!Objects.equals(systemUser, activeUser.getShortName()) && !userUtils.isSuperUser(activeUser)) {
+        if (activeUser == null || !isSystemOrSuperUser(activeUser)) {
             throw new AccessDeniedException("User '" + user.getShortName() + "is not system or super user.");
         }
     }
@@ -1390,7 +1407,7 @@ boolean canSkipAccessCheck(User user, final String operation, String access, fin
             LOG.warn("canSkipAccessCheck: exiting{}", "Unexpeceted: User is null: access denied, not audited!");
 
             throw new AccessDeniedException("No user associated with request (" + operation + ") for action: " + access + "on table:" + table);
-        } else if (isAccessForMetadataRead(access, table)) {
+        } else if (isAccessForMetadataRead(access, table, user)) {
             LOG.debug("canSkipAccessCheck: true: metadata read access always allowed, not audited");
 
             result = true;
@@ -1434,9 +1451,15 @@ boolean canSkipAccessCheck(User user, final String operation, String access, fin
 
     /* ---- EndpointObserver implementation ---- */
 
-    boolean isAccessForMetadataRead(String access, String table) {
+    boolean isAccessForMetadataRead(String access, String table, User user) {
         if (authUtils.isReadAccess(access) && isSpecialTable(table)) {
-            LOG.debug("isAccessForMetadataRead: Metadata tables read: access allowed!");
+            if (StringUtils.equals(table, "hbase:acl")) {
+                if (!isSystemOrSuperUser(user)) {
+                    LOG.debug("isAccessForMetadataRead: Metadata tables read: not access allowed for user: {}!", (user != null ? user.getShortName() : ""));
+                    return false;
+                }
+            }
+            LOG.debug("isAccessForMetadataRead: Metadata tables read: access allowed for user: {}!", (user != null ? user.getShortName() : ""));
 
             return true;
         }

hbase-agent/src/test/java/org/apache/ranger/authorization/hbase/RangerAuthorizationCoprocessorTest.java

@@ -391,12 +391,36 @@ public void test18_requirePermission_region_allowsOrDenies() throws Exception {
     }
 
     @Test
-    public void test19_isSpecialTable_and_metadataRead() {
+    public void test19_isSpecialTable_and_metadataRead() throws Exception {
         RangerAuthorizationCoprocessor cp = new RangerAuthorizationCoprocessor();
         Assertions.assertTrue(cp.isSpecialTable("hbase:meta"));
         Assertions.assertFalse(cp.isSpecialTable("normal"));
-        Assertions.assertTrue(cp.isAccessForMetadataRead("read", "hbase:acl"));
-        Assertions.assertFalse(cp.isAccessForMetadataRead("write", "hbase:acl"));
+        Assertions.assertFalse(cp.isAccessForMetadataRead("read", "hbase:acl", null));
+        Assertions.assertFalse(cp.isAccessForMetadataRead("write", "hbase:acl", null));
+
+        // Test for system user bypass on hbase:acl
+        User systemUser = mock(User.class);
+        when(systemUser.getShortName()).thenReturn(User.getCurrent().getShortName());
+        Assertions.assertTrue(cp.isAccessForMetadataRead("read", "hbase:acl", systemUser));
+
+        // Test for super user bypass on hbase:acl
+        User superUser = mock(User.class);
+        when(superUser.getShortName()).thenReturn("some_super_user");
+        HbaseUserUtils userUtils = mock(HbaseUserUtils.class);
+        lenient().when(userUtils.isSuperUser(superUser)).thenReturn(true);
+        Field userUtilsField = RangerAuthorizationCoprocessor.class.getDeclaredField("userUtils");
+        userUtilsField.setAccessible(true);
+        userUtilsField.set(cp, userUtils);
+        Assertions.assertTrue(cp.isAccessForMetadataRead("read", "hbase:acl", superUser));
+
+        // Test for normal user on hbase:acl (should be denied)
+        User normalUser = mock(User.class);
+        when(normalUser.getShortName()).thenReturn("normal_user");
+        lenient().when(userUtils.isSuperUser(normalUser)).thenReturn(false);
+        Assertions.assertFalse(cp.isAccessForMetadataRead("read", "hbase:acl", normalUser));
+
+        // Test for normal user on hbase:meta (should be allowed)
+        Assertions.assertTrue(cp.isAccessForMetadataRead("read", "hbase:meta", normalUser));
     }
 
     @Test