Allow only root user to read fate and scanref tables by default (#6203)

Allow only root user to read fate and scanref tables by default

Closes #6137

Author: dlmarion

server/base/src/main/java/org/apache/accumulo/server/security/SecurityOperation.java

@@ -31,7 +31,6 @@
 import org.apache.accumulo.core.client.TableNotFoundException;
 import org.apache.accumulo.core.client.security.tokens.AuthenticationToken;
 import org.apache.accumulo.core.clientImpl.Credentials;
-import org.apache.accumulo.core.clientImpl.Namespace;
 import org.apache.accumulo.core.clientImpl.thrift.SecurityErrorCode;
 import org.apache.accumulo.core.clientImpl.thrift.ThriftSecurityException;
 import org.apache.accumulo.core.conf.Property;
@@ -355,6 +354,7 @@ private boolean _hasTablePermission(String user, TableId table, TablePermission
       boolean useCached) throws ThriftSecurityException {
     targetUserExists(user);
 
+    // Allow all users to read root and metadata tables
     if ((table.equals(SystemTables.METADATA.tableId()) || table.equals(SystemTables.ROOT.tableId()))
         && permission.equals(TablePermission.READ)) {
       return true;
@@ -384,10 +384,6 @@ private boolean _hasNamespacePermission(String user, NamespaceId namespace,
 
     targetUserExists(user);
 
-    if (namespace.equals(Namespace.ACCUMULO.id()) && permission.equals(NamespacePermission.READ)) {
-      return true;
-    }
-
     try {
       if (useCached) {
         return permHandle.hasCachedNamespacePermission(user, namespace.canonical(), permission);

test/src/main/java/org/apache/accumulo/test/functional/PermissionsIT.java

@@ -22,6 +22,7 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assumptions.assumeTrue;
 
@@ -694,6 +695,42 @@ public void tablePermissionTest() throws Exception {
     }
     loginAs(rootUser);
     try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+
+      // check root user permissions on FATE and SCAN_REF tables
+      verifyHasOnlyTheseTablePermissions(c, c.whoami(), SystemTables.FATE.tableName());
+      verifyHasOnlyTheseTablePermissions(c, c.whoami(), SystemTables.SCAN_REF.tableName());
+      for (String tn : SystemTables.tableNames()) {
+        try (Scanner s = c.createScanner(tn)) {
+          if (SystemTables.ROOT.tableName().equals(tn)
+              || SystemTables.METADATA.tableName().equals(tn)) {
+            @SuppressWarnings("unused")
+            var unused = s.iterator().hasNext();
+          } else {
+            IllegalStateException ise =
+                assertThrows(IllegalStateException.class, () -> s.iterator().hasNext());
+            assertTrue(ise.getMessage().contains("Error PERMISSION_DENIED for user"),
+                "Permission denied error not thrown for root user scan of table: " + tn
+                    + ". Message = " + ise.getMessage());
+          }
+        }
+      }
+
+      // Give the root user the read permission and try again.
+      c.securityOperations().grantTablePermission(rootUser.getPrincipal(),
+          SystemTables.FATE.tableName(), TablePermission.READ);
+      c.securityOperations().grantTablePermission(rootUser.getPrincipal(),
+          SystemTables.SCAN_REF.tableName(), TablePermission.READ);
+      verifyHasOnlyTheseTablePermissions(c, c.whoami(), SystemTables.FATE.tableName(),
+          TablePermission.READ);
+      verifyHasOnlyTheseTablePermissions(c, c.whoami(), SystemTables.SCAN_REF.tableName(),
+          TablePermission.READ);
+      for (String tn : SystemTables.tableNames()) {
+        try (Scanner s = c.createScanner(tn)) {
+          @SuppressWarnings("unused")
+          var unused = s.iterator().hasNext();
+        }
+      }
+
       c.securityOperations().createLocalUser(principal, passwordToken);
       loginAs(testUser);
       try (AccumuloClient test_user_client =
@@ -703,8 +740,31 @@ public void tablePermissionTest() throws Exception {
         loginAs(rootUser);
         verifyHasOnlyTheseTablePermissions(c, c.whoami(), SystemTables.METADATA.tableName(),
             TablePermission.READ, TablePermission.ALTER_TABLE);
-        String tableName = getUniqueNames(1)[0] + "__TABLE_PERMISSION_TEST__";
 
+        // check test user permissions on FATE and SCAN_REF tables
+        loginAs(testUser);
+        verifyHasOnlyTheseTablePermissions(c, test_user_client.whoami(),
+            SystemTables.FATE.tableName());
+        verifyHasOnlyTheseTablePermissions(c, test_user_client.whoami(),
+            SystemTables.SCAN_REF.tableName());
+        for (String tn : SystemTables.tableNames()) {
+          try (Scanner s = test_user_client.createScanner(tn)) {
+            if (SystemTables.ROOT.tableName().equals(tn)
+                || SystemTables.METADATA.tableName().equals(tn)) {
+              @SuppressWarnings("unused")
+              var unused = s.iterator().hasNext();
+            } else {
+              IllegalStateException ise =
+                  assertThrows(IllegalStateException.class, () -> s.iterator().hasNext());
+              assertTrue(ise.getMessage().contains("Error PERMISSION_DENIED for user"),
+                  "Permission denied error not thrown for root user scan of table: " + tn
+                      + ". Message = " + ise.getMessage());
+            }
+          }
+        }
+
+        loginAs(rootUser);
+        String tableName = getUniqueNames(1)[0] + "__TABLE_PERMISSION_TEST__";
         // test each permission
         for (TablePermission perm : TablePermission.values()) {
           if (perm == TablePermission.ALTER_TABLE) {