DISCUSS: Don't modify the root user on table create

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

@@ -342,7 +342,8 @@ private boolean _hasSystemPermission(String user, SystemPermission permission, b
   private boolean hasTablePermission(TCredentials credentials, TableId tableId,
       NamespaceId namespaceId, TablePermission permission, boolean useCached)
       throws ThriftSecurityException {
-    if (isSystemUser(credentials)) {
+    if (isSystemUser(credentials)
+        || hasSystemPermission(credentials, SystemPermission.SYSTEM, false)) {
       return true;
     }
     return _hasTablePermission(credentials.getPrincipal(), tableId, permission, useCached)

server/manager/src/main/java/org/apache/accumulo/manager/tableOps/create/SetupPermissions.java

@@ -20,6 +20,7 @@
 
 import org.apache.accumulo.core.clientImpl.thrift.ThriftSecurityException;
 import org.apache.accumulo.core.fate.Repo;
+import org.apache.accumulo.core.security.SystemPermission;
 import org.apache.accumulo.core.security.TablePermission;
 import org.apache.accumulo.manager.Manager;
 import org.apache.accumulo.manager.tableOps.ManagerRepo;
@@ -38,9 +39,12 @@ class SetupPermissions extends ManagerRepo {
 
   @Override
   public Repo<Manager> call(long tid, Manager env) throws Exception {
-    // give all table permissions to the creator
+    // give all table permissions to the creator if that creator is not the system user or has
+    // SYSTEM level permissions
     var security = env.getContext().getSecurityOperation();
-    if (!tableInfo.getUser().equals(env.getContext().getCredentials().getPrincipal())) {
+    if (!tableInfo.getUser().equals(env.getContext().getCredentials().getPrincipal())
+        && !security.hasSystemPermission(env.getContext().rpcCreds(), tableInfo.getUser(),
+            SystemPermission.SYSTEM)) {
       for (TablePermission permission : TablePermission.values()) {
         try {
           security.grantTablePermission(env.getContext().rpcCreds(), tableInfo.getUser(),

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

@@ -47,12 +47,14 @@
 import org.apache.accumulo.core.client.security.tokens.AuthenticationToken;
 import org.apache.accumulo.core.client.security.tokens.PasswordToken;
 import org.apache.accumulo.core.client.summary.Summary;
+import org.apache.accumulo.core.clientImpl.Namespace;
 import org.apache.accumulo.core.conf.Property;
 import org.apache.accumulo.core.data.Key;
 import org.apache.accumulo.core.data.Mutation;
 import org.apache.accumulo.core.data.Value;
 import org.apache.accumulo.core.metadata.MetadataTable;
 import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.security.NamespacePermission;
 import org.apache.accumulo.core.security.SystemPermission;
 import org.apache.accumulo.core.security.TablePermission;
 import org.apache.accumulo.harness.AccumuloClusterHarness;
@@ -725,6 +727,57 @@ public void tablePermissionTest() throws Exception {
     }
   }
 
+  @Test
+  public void rootUserTablePermissionTest() throws Exception {
+    // create the test user
+    ClusterUser testUser = getUser(0);
+    ClusterUser rootUser = getAdminUser();
+
+    String principal = testUser.getPrincipal();
+    AuthenticationToken token = testUser.getToken();
+    PasswordToken passwordToken = null;
+    if (token instanceof PasswordToken) {
+      passwordToken = (PasswordToken) token;
+    }
+    loginAs(rootUser);
+    try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+      c.securityOperations().createLocalUser(principal, passwordToken);
+      loginAs(testUser);
+      try (AccumuloClient test_user_client =
+          Accumulo.newClient().from(c.properties()).as(principal, token).build()) {
+
+        String tableName = getUniqueNames(1)[0] + "__TABLE_READ_PERMISSION_TEST__";
+        // Allow test user to create a table in default namespace
+        loginAs(rootUser);
+        c.securityOperations().grantNamespacePermission(testUser.getPrincipal(),
+            Namespace.DEFAULT.name(), NamespacePermission.CREATE_TABLE);
+
+        // create the test table
+        test_user_client.tableOperations().create(tableName);
+        // put in some initial data
+        try (BatchWriter writer = test_user_client.createBatchWriter(tableName)) {
+          Mutation m = new Mutation(new Text("row"));
+          m.put("cf", "cq", "val");
+          writer.addMutation(m);
+        }
+
+        // Attempt to scan table as test user
+        try (Scanner scanner = test_user_client.createScanner(tableName, Authorizations.EMPTY)) {
+          for (Entry<Key,Value> keyValueEntry : scanner) {
+            assertNotNull(keyValueEntry);
+          }
+        }
+
+        // Attempt to scan table as root user
+        try (Scanner scanner = c.createScanner(tableName, Authorizations.EMPTY)) {
+          for (Entry<Key,Value> keyValueEntry : scanner) {
+            assertNotNull(keyValueEntry);
+          }
+        }
+      }
+    }
+  }
+
   private void createTestTable(AccumuloClient c, String testUser, String tableName)
       throws Exception {
     if (!c.tableOperations().exists(tableName)) {