Added ScanServer property for allowed tables

core/src/main/java/org/apache/accumulo/core/conf/Property.java

@@ -561,6 +561,13 @@ public enum Property {
       "The amount of time a scan reference is unused before its deleted from metadata table.",
       "2.1.0"),
   @Experimental
+  SSERV_SCAN_ALLOWED_TABLES("sserver.scan.allowed.tables.group.", null, PropertyType.PREFIX,
+      "A regular expression that determines which tables are allowed to be scanned for"
+          + " servers in the specified group. The property name should end with the scan server"
+          + " group and the property value should take into account the table namespace and name."
+          + " The default value disallows scans on tables in the accumulo namespace.",
+      "2.1.5"),
+  @Experimental
   SSERV_THREADCHECK("sserver.server.threadcheck.time", "1s", PropertyType.TIMEDURATION,
       "The time between adjustments of the thrift server thread pool.", "2.1.0"),
   // properties that are specific to tablet server behavior

server/tserver/src/main/java/org/apache/accumulo/tserver/ScanServer.java

@@ -45,15 +45,20 @@
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.regex.PatternSyntaxException;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 import org.apache.accumulo.core.Constants;
 import org.apache.accumulo.core.client.AccumuloException;
+import org.apache.accumulo.core.clientImpl.thrift.SecurityErrorCode;
 import org.apache.accumulo.core.clientImpl.thrift.ThriftSecurityException;
 import org.apache.accumulo.core.conf.AccumuloConfiguration;
 import org.apache.accumulo.core.conf.Property;
 import org.apache.accumulo.core.conf.cluster.ClusterConfigParser;
+import org.apache.accumulo.core.data.TableId;
 import org.apache.accumulo.core.dataImpl.KeyExtent;
 import org.apache.accumulo.core.dataImpl.thrift.InitialMultiScan;
 import org.apache.accumulo.core.dataImpl.thrift.InitialScan;
@@ -177,6 +182,8 @@ private TabletMetadataLoader(Ample ample) {
   }
 
   private static final Logger LOG = LoggerFactory.getLogger(ScanServer.class);
+  // Default pattern to allow scans on all tables not in accumulo namespace
+  private static final String DEFAULT_SCAN_ALLOWED_PATTERN = "^(?!accumulo\\.).*$";
 
   protected ThriftScanClientHandler delegate;
   private UUID serverLockUUID;
@@ -213,6 +220,9 @@ private TabletMetadataLoader(Ample ample) {
 
   private final String groupName;
 
+  private final ConcurrentHashMap<TableId,Boolean> allowedTables = new ConcurrentHashMap<>();
+  private volatile String currentAllowedTableRegex;
+
   public ScanServer(ScanServerOpts opts, String[] args) {
     super("sserver", opts, args);
 
@@ -388,6 +398,7 @@ public void run() {
     }
 
     SecurityUtil.serverLogin(getConfiguration());
+    updateAllowedTables(false);
 
     ServerAddress address = null;
     try {
@@ -423,6 +434,7 @@ public void run() {
           Thread.sleep(1000);
           updateIdleStatus(sessionManager.getActiveScans().isEmpty()
               && tabletMetadataCache.estimatedSize() == 0);
+          updateAllowedTables(false);
         } catch (InterruptedException e) {
           LOG.info("Interrupt Exception received, shutting down");
           gracefulShutdown(getContext().rpcCreds());
@@ -477,6 +489,86 @@ public void run() {
     }
   }
 
+  // Visible for testing
+  protected boolean isAllowed(TCredentials credentials, TableId tid)
+      throws ThriftSecurityException {
+    Boolean result = allowedTables.get(tid);
+    while (result == null) {
+      LOG.debug(
+          "Allowed tables mapping does not contain an entry for table: {}, refreshing table...",
+          tid);
+      // Clear the cache and try again, maybe there
+      // is a race condition in table creation and scan
+      updateAllowedTables(true);
+      // validate that the table exists, else throw
+      delegate.getNamespaceId(credentials, tid);
+      result = allowedTables.get(tid);
+    }
+    return result;
+  }
+
+  private synchronized void updateAllowedTables(boolean clearCache) {
+
+    LOG.debug("Updating allowed tables for ScanServer");
+    if (clearCache) {
+      context.clearTableListCache();
+    }
+
+    // Remove tables that no longer exist
+    allowedTables.keySet().forEach(tid -> {
+      if (!getContext().getTableIdToNameMap().containsKey(tid)) {
+        LOG.debug("Removing table {} from allowed table map as it no longer exists", tid);
+        allowedTables.remove(tid);
+      }
+    });
+
+    final String propName = Property.SSERV_SCAN_ALLOWED_TABLES.getKey() + groupName;
+    String allowedTableRegex = getConfiguration().get(propName);
+    if (allowedTableRegex == null) {
+      allowedTableRegex = DEFAULT_SCAN_ALLOWED_PATTERN;
+    }
+
+    if (currentAllowedTableRegex == null) {
+      LOG.debug("Property {} initial value: {}", propName, allowedTableRegex);
+    } else if (currentAllowedTableRegex.equals(allowedTableRegex)) {
+      // Property value has not changed, do nothing
+    } else {
+      LOG.debug("Property {} has changed. Old value: {}, new value: {}", propName,
+          currentAllowedTableRegex, allowedTableRegex);
+    }
+
+    Pattern allowedTablePattern;
+    try {
+      allowedTablePattern = Pattern.compile(allowedTableRegex);
+      // Regex is valid, store it
+      currentAllowedTableRegex = allowedTableRegex;
+    } catch (PatternSyntaxException e) {
+      LOG.error(
+          "Property {} contains an invalid regular expression. Property value: {}. Disabling all tables.",
+          propName, allowedTableRegex);
+      allowedTablePattern = null;
+    }
+
+    Pattern p = allowedTablePattern;
+    context.getTableNameToIdMap().entrySet().forEach(e -> {
+      String tname = e.getKey();
+      TableId tid = e.getValue();
+      if (p == null) {
+        allowedTables.put(tid, Boolean.FALSE);
+      } else {
+        Matcher m = p.matcher(tname);
+        if (m.matches()) {
+          LOG.debug("Table {} can now be scanned via this ScanServer", tname);
+          allowedTables.put(tid, Boolean.TRUE);
+        } else {
+          LOG.debug("Table {} cannot be scanned via this ScanServer", tname);
+          allowedTables.put(tid, Boolean.FALSE);
+        }
+      }
+    });
+
+  }
+
   @SuppressWarnings("unchecked")
   private Map<KeyExtent,TabletMetadata> getTabletMetadata(Collection<KeyExtent> extents) {
     if (tabletMetadataCache == null) {
@@ -945,11 +1037,6 @@ public void close() {
     };
   }
 
-  /* Exposed for testing */
-  protected boolean isSystemUser(TCredentials creds) {
-    return context.getSecurityOperation().isSystemUser(creds);
-  }
-
   @Override
   public InitialScan startScan(TInfo tinfo, TCredentials credentials, TKeyExtent textent,
       TRange range, List<TColumn> columns, int batchSize, List<IterInfo> ssiList,
@@ -966,9 +1053,9 @@ public InitialScan startScan(TInfo tinfo, TCredentials credentials, TKeyExtent t
 
     KeyExtent extent = getKeyExtent(textent);
 
-    if (extent.isMeta() && !isSystemUser(credentials)) {
-      throw new TException(
-          "Only the system user can perform eventual consistency scans on the root and metadata tables");
+    if (!isAllowed(credentials, extent.tableId())) {
+      throw new TException("Scan of table " + extent.tableId() + " disallowed by property: "
+          + Property.SSERV_SCAN_ALLOWED_TABLES.getKey() + this.groupName);
     }
 
     try (ScanReservation reservation =
@@ -1038,9 +1125,11 @@ public InitialMultiScan startMultiScan(TInfo tinfo, TCredentials credentials,
     for (Entry<TKeyExtent,List<TRange>> entry : tbatch.entrySet()) {
       KeyExtent extent = getKeyExtent(entry.getKey());
 
-      if (extent.isMeta() && !context.getSecurityOperation().isSystemUser(credentials)) {
-        throw new TException(
-            "Only the system user can perform eventual consistency scans on the root and metadata tables");
+      if (!isAllowed(credentials, extent.tableId())) {
+        throw new ThriftSecurityException(
+            "Scan of table " + extent.tableId() + " disallowed by property: "
+                + Property.SSERV_SCAN_ALLOWED_TABLES.getKey() + this.groupName,
+            SecurityErrorCode.PERMISSION_DENIED);
       }
 
       batch.put(extent, entry.getValue());

server/tserver/src/main/java/org/apache/accumulo/tserver/ThriftScanClientHandler.java

@@ -109,7 +109,7 @@ public ThriftScanClientHandler(TabletHostingServer server, WriteTracker writeTra
         .getTimeInMillis(Property.TSERV_SCAN_RESULTS_MAX_TIMEOUT);
   }
 
-  private NamespaceId getNamespaceId(TCredentials credentials, TableId tableId)
+  public NamespaceId getNamespaceId(TCredentials credentials, TableId tableId)
       throws ThriftSecurityException {
     try {
       return server.getContext().getNamespaceId(tableId);