Added property to restrict bulk import source file locations

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

@@ -313,6 +313,10 @@ public enum Property {
           + " user-implementations of pluggable Accumulo features, such as the balancer"
           + " or volume chooser.",
       "2.0.0"),
+  GENERAL_BULK_SOURCE_DIRS("general.bulk.user.allowed.source.dir", "{}", PropertyType.JSON,
+      "Map of Accumulo user name to Java regular expression, where the regular expression "
+          + "will specify the allowed source location of bulk import files.",
+      "2.1.7"),
   GENERAL_CACHE_MANAGER_IMPL("general.block.cache.manager.class",
       "org.apache.accumulo.core.file.blockfile.cache.lru.LruBlockCacheManager", PropertyType.STRING,
       "Specifies the class name of the block cache factory implementation."
@@ -2126,6 +2130,7 @@ public static boolean isValidZooPropertyKey(String key) {
         || key.startsWith(Property.MASTER_PREFIX.getKey())
         || key.startsWith(Property.GC_PREFIX.getKey())
         || key.startsWith(Property.GENERAL_ARBITRARY_PROP_PREFIX.getKey())
+        || key.equals(Property.GENERAL_BULK_SOURCE_DIRS.getKey())
         || key.equals(Property.GENERAL_FILE_NAME_ALLOCATION_BATCH_SIZE_MIN.getKey())
         || key.equals(Property.GENERAL_FILE_NAME_ALLOCATION_BATCH_SIZE_MAX.getKey())
         || key.equals(Property.GENERAL_SERVER_ITERATOR_OPTIONS_COMPRESSION_ALGO.getKey())

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

@@ -20,10 +20,13 @@
 
 import static java.nio.charset.StandardCharsets.UTF_8;
 
+import java.lang.reflect.Type;
 import java.nio.ByteBuffer;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.function.Supplier;
+import java.util.regex.Pattern;
 
 import org.apache.accumulo.core.Constants;
 import org.apache.accumulo.core.client.AccumuloSecurityException;
@@ -59,6 +62,10 @@
 import org.slf4j.LoggerFactory;
 
 import com.google.common.base.Suppliers;
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import com.google.gson.JsonSyntaxException;
+import com.google.gson.reflect.TypeToken;
 
 /**
  * Utility class for performing various security operations with the appropriate checks
@@ -73,6 +80,8 @@ public class SecurityOperation {
   private final Supplier<String> rootUserName;
   private final ZooCache zooCache;
   private final String zkUserPath;
+  private final Gson gson = new GsonBuilder().create();
+  private final Type jsonMapType = new TypeToken<Map<String,String>>() {}.getType();
 
   protected final ServerContext context;
 
@@ -537,7 +546,33 @@ protected boolean canDeleteRange(TCredentials c, TableId tableId, String tableNa
   protected boolean canBulkImport(TCredentials c, TableId tableId, String tableName, String dir,
       String failDir, NamespaceId namespaceId) throws ThriftSecurityException {
     authenticate(c);
-    return hasTablePermission(c, tableId, namespaceId, TablePermission.BULK_IMPORT, false);
+    final boolean tablePerm =
+        hasTablePermission(c, tableId, namespaceId, TablePermission.BULK_IMPORT, false);
+
+    final String user = c.getPrincipal();
+    final String allowedLocations =
+        context.getConfiguration().get(Property.GENERAL_BULK_SOURCE_DIRS);
+    boolean locationAllowed = false;
+    try {
+      final Map<String,String> allowedLocationsMap = gson.fromJson(allowedLocations, jsonMapType);
+      final String userSourceLocationPattern = allowedLocationsMap.get(user);
+      if (userSourceLocationPattern == null) {
+        log.error("User " + user + " does not have an allowed source dir location in property "
+            + Property.GENERAL_BULK_SOURCE_DIRS.getKey());
+      } else if (Pattern.matches(userSourceLocationPattern, dir)) {
+        locationAllowed = true;
+      }
+    } catch (JsonSyntaxException e) {
+      log.error(
+          "Property " + Property.GENERAL_BULK_SOURCE_DIRS.getKey() + " value is not valid json", e);
+    }
+
+    if (!locationAllowed) {
+      log.error(
+          "Bulk Import into table {} by user {} not allowed from directory {}, check table property {} value: {}",
+          tableId, user, dir, Property.GENERAL_BULK_SOURCE_DIRS.getKey(), allowedLocations);
+    }
+    return tablePerm && locationAllowed;
   }
 
   protected boolean canCompact(TCredentials c, TableId tableId, NamespaceId namespaceId)

test/src/main/java/org/apache/accumulo/test/AuditMessageIT.java

@@ -56,6 +56,7 @@
 import org.apache.accumulo.core.security.TablePermission;
 import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
 import org.apache.accumulo.server.security.AuditedSecurityOperation;
+import org.apache.accumulo.test.functional.BulkIT;
 import org.apache.accumulo.test.functional.ConfigurableMacBase;
 import org.apache.commons.io.FileUtils;
 import org.apache.hadoop.io.Text;
@@ -350,6 +351,9 @@ public void testImportExportOperationsAudits() throws AccumuloSecurityException,
         Collections.singleton(exportDir.toString()), ImportConfiguration.empty());
 
     // Now do a Directory (bulk) import of the same data.
+    BulkIT.setBulkSourceLocations(auditAccumuloClient, getCluster().getFileSystem(),
+        exportDirBulk.getParentFile());
+
     auditAccumuloClient.tableOperations().create(THIRD_TEST_TABLE_NAME);
     auditAccumuloClient.tableOperations().importDirectory(exportDirBulk.toString())
         .to(THIRD_TEST_TABLE_NAME).load();

test/src/main/java/org/apache/accumulo/test/BulkImportMonitoringIT.java

@@ -46,6 +46,7 @@
 import org.apache.accumulo.core.util.Pair;
 import org.apache.accumulo.minicluster.ServerType;
 import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
+import org.apache.accumulo.test.functional.BulkIT;
 import org.apache.accumulo.test.functional.ConfigurableMacBase;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
@@ -95,6 +96,8 @@ public void test() throws Exception {
       fs.delete(base, true);
       fs.mkdirs(base);
 
+      BulkIT.setBulkSourceLocations(c, fs, base);
+
       ExecutorService es = Executors.newFixedThreadPool(5);
       List<Future<Pair<String,String>>> futures = new ArrayList<>();
       for (int i = 0; i < 10; i++) {

test/src/main/java/org/apache/accumulo/test/BulkImportSequentialRowsIT.java

@@ -30,6 +30,7 @@
 import org.apache.accumulo.core.security.Authorizations;
 import org.apache.accumulo.harness.AccumuloClusterHarness;
 import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
+import org.apache.accumulo.test.functional.BulkIT;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FsShell;
@@ -103,6 +104,7 @@ public void testBulkImportFailure() throws Exception {
       // because
       // of the balancer moving tablets around
       // and then we get to verify that the bug is actually fixed.
+      BulkIT.setBulkSourceLocations(client, fs, bulk.getParent());
       to.importDirectory(bulk.toString()).to(tableName).load();
 
       // The bug is that some tablets don't get imported into.

test/src/main/java/org/apache/accumulo/test/BulkImportVolumeIT.java

@@ -29,6 +29,7 @@
 import org.apache.accumulo.core.conf.Property;
 import org.apache.accumulo.harness.AccumuloClusterHarness;
 import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
+import org.apache.accumulo.test.functional.BulkIT;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
@@ -96,6 +97,7 @@ public void testBulkImportFailure() throws Exception {
       log.info("bogus: {}", bogus);
       assertTrue(fs.exists(bogus));
       log.info("Importing {} into {} with failures directory {}", bulk, tableName, err);
+      BulkIT.setBulkSourceLocations(client, fs, bulk.getParent());
       client.tableOperations().importDirectory(tableName, bulk.toString(), err.toString(), false);
       assertEquals(1, fs.listStatus(err).length);
     }

test/src/main/java/org/apache/accumulo/test/CountNameNodeOpsBulkIT.java

@@ -47,6 +47,7 @@
 import org.apache.accumulo.core.spi.crypto.NoCryptoServiceFactory;
 import org.apache.accumulo.minicluster.ServerType;
 import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
+import org.apache.accumulo.test.functional.BulkIT;
 import org.apache.accumulo.test.functional.ConfigurableMacBase;
 import org.apache.accumulo.test.functional.FunctionalTestUtils;
 import org.apache.hadoop.conf.Configuration;
@@ -122,6 +123,7 @@ public void compareOldNewBulkImportTest() throws Exception {
       final Path base = new Path(basePath, "testBulkLoad" + tableName);
       fs.delete(base, true);
       fs.mkdirs(base);
+      BulkIT.setBulkSourceLocations(c, fs, base);
 
       ExecutorService es = Executors.newFixedThreadPool(5);
       List<Future<String>> futures = new ArrayList<>();

test/src/main/java/org/apache/accumulo/test/ScanConsistencyIT.java

@@ -58,6 +58,7 @@
 import org.apache.accumulo.core.iterators.IteratorEnvironment;
 import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
 import org.apache.accumulo.harness.AccumuloClusterHarness;
+import org.apache.accumulo.test.functional.BulkIT;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
@@ -109,6 +110,7 @@ public void testConcurrentScanConsistency() throws Exception {
 
       TestContext testContext = new TestContext(client, table, getCluster().getFileSystem(),
           getCluster().getTemporaryPath().toString());
+      BulkIT.setBulkSourceLocations(client, getFileSystem(), getCluster().getTemporaryPath());
 
       List<Future<WriteStats>> writeTasks = new ArrayList<>();
       List<Future<ScanStats>> scanTasks = new ArrayList<>();

test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompactionProgressIT.java

@@ -30,6 +30,7 @@
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import java.io.File;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.EnumSet;
@@ -61,6 +62,7 @@
 import org.apache.accumulo.harness.AccumuloClusterHarness;
 import org.apache.accumulo.minicluster.ServerType;
 import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
+import org.apache.accumulo.test.functional.BulkIT;
 import org.apache.accumulo.test.functional.SlowIterator;
 import org.apache.accumulo.test.metrics.TestStatsDRegistryFactory;
 import org.apache.accumulo.test.metrics.TestStatsDSink;
@@ -378,6 +380,7 @@ public void testProgressWithBulkImport() throws Exception {
       String dir = getDir(client, tableName1);
 
       log.info("Bulk importing files in dir " + dir + " to table " + tableName2);
+      BulkIT.setBulkSourceLocations(client, getFileSystem(), new File(dir).getParentFile());
       client.tableOperations().importDirectory(dir).to(tableName2).load();
       log.info("Finished bulk import");
 

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

@@ -131,6 +131,7 @@ protected void runTest(String table, long fateTxid, Loader loader) throws IOExce
       // determine the files new name and path
       FileStatus status = fs.listStatus(new Path(bulkDir))[0];
       Path bulkLoadPath = fs.makeQualified(status.getPath());
+      BulkIT.setBulkSourceLocations(c, fs, bulkLoadPath);
 
       // Directly ask the tablet to load the file.
       loader.load(fateTxid, asCtx, extent, bulkLoadPath, status.getLen(), false);

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

@@ -19,31 +19,44 @@
 package org.apache.accumulo.test.functional;
 
 import static org.apache.accumulo.harness.AccumuloITBase.SUNNY_DAY;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import java.io.File;
 import java.io.IOException;
 import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
 
 import org.apache.accumulo.core.client.Accumulo;
 import org.apache.accumulo.core.client.AccumuloClient;
 import org.apache.accumulo.core.client.AccumuloException;
 import org.apache.accumulo.core.client.AccumuloSecurityException;
 import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.conf.Property;
 import org.apache.accumulo.harness.AccumuloClusterHarness;
 import org.apache.accumulo.test.TestIngest;
 import org.apache.accumulo.test.TestIngest.IngestParams;
 import org.apache.accumulo.test.VerifyIngest;
 import org.apache.accumulo.test.VerifyIngest.VerifyParams;
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.gson.GsonBuilder;
 
 /**
  * Tests Old and New Bulk import
  */
 @Tag(SUNNY_DAY)
 public class BulkIT extends AccumuloClusterHarness {
 
+  private static final Logger LOG = LoggerFactory.getLogger(BulkIT.class);
+
   private static final int N = 100000;
   private static final int COUNT = 5;
 
@@ -52,19 +65,68 @@ protected Duration defaultTimeout() {
     return Duration.ofMinutes(4);
   }
 
+  public static void setBulkSourceLocations(AccumuloClient client, FileSystem fs,
+      File localDirectory)
+      throws AccumuloException, AccumuloSecurityException, IllegalArgumentException, IOException {
+    Path p = FileSystem.get(localDirectory.toURI(), new Configuration())
+        .makeQualified(new Path(localDirectory.toString()));
+    setBulkSourceLocations(client, fs, p);
+  }
+
+  public static void setBulkSourceLocations(AccumuloClient client, FileSystem fs, Path basePath)
+      throws AccumuloException, AccumuloSecurityException {
+    Map<String,String> userPatterns = new HashMap<>();
+    userPatterns.put(client.whoami(), fs.makeQualified(basePath).toString() + "/.*");
+    String propVal = new GsonBuilder().create().toJson(userPatterns);
+    LOG.info("Property set to {}", propVal);
+    client.instanceOperations().setProperty(Property.GENERAL_BULK_SOURCE_DIRS.getKey(), propVal);
+  }
+
   @Test
   public void test() throws Exception {
     try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
-      runTest(client, getCluster().getFileSystem(), getCluster().getTemporaryPath(),
-          getUniqueNames(1)[0], this.getClass().getName(), testName(), false);
+      Path basePath = getCluster().getTemporaryPath();
+      FileSystem fs = getCluster().getFileSystem();
+      setBulkSourceLocations(client, fs, basePath);
+      runTest(client, fs, basePath, getUniqueNames(1)[0], this.getClass().getName(), testName(),
+          false);
     }
   }
 
   @Test
   public void testOld() throws Exception {
     try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
-      runTest(client, getCluster().getFileSystem(), getCluster().getTemporaryPath(),
-          getUniqueNames(1)[0], this.getClass().getName(), testName(), true);
+      Path basePath = getCluster().getTemporaryPath();
+      FileSystem fs = getCluster().getFileSystem();
+      setBulkSourceLocations(client, fs, basePath);
+      runTest(client, fs, basePath, getUniqueNames(1)[0], this.getClass().getName(), testName(),
+          true);
+    }
+  }
+
+  @Test
+  public void testWithoutSourceDirProperty() throws Exception {
+    try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+      Path basePath = getCluster().getTemporaryPath();
+      FileSystem fs = getCluster().getFileSystem();
+      AccumuloSecurityException ase =
+          assertThrows(AccumuloSecurityException.class, () -> runTest(client, fs, basePath,
+              getUniqueNames(1)[0], this.getClass().getName(), testName(), false));
+      assertTrue(ase.getMessage().startsWith("Error PERMISSION_DENIED for user root on table "));
+      assertTrue(ase.getMessage().endsWith("User does not have permission to perform this action"));
+    }
+  }
+
+  @Test
+  public void testOldWithoutSourceDirProperty() throws Exception {
+    try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+      Path basePath = getCluster().getTemporaryPath();
+      FileSystem fs = getCluster().getFileSystem();
+      AccumuloSecurityException ase =
+          assertThrows(AccumuloSecurityException.class, () -> runTest(client, fs, basePath,
+              getUniqueNames(1)[0], this.getClass().getName(), testName(), true));
+      assertTrue(ase.getMessage().startsWith("Error PERMISSION_DENIED for user root on table "));
+      assertTrue(ase.getMessage().endsWith("User does not have permission to perform this action"));
     }
   }
 

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

@@ -164,6 +164,8 @@ public void setupBulkTest() throws Exception {
       aconf = getCluster().getServerContext().getConfiguration();
       fs = getCluster().getFileSystem();
       rootPath = getCluster().getTemporaryPath().toString();
+      BulkIT.setBulkSourceLocations(c, fs, getCluster().getTemporaryPath());
+
     }
   }
 

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

@@ -105,7 +105,10 @@ public static String writeNonContiguousData(FileSystem fs, String file,
   public void setupCluster() throws Exception {
     super.setupCluster();
     // prime the zk connection
-    try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {}
+    try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+      BulkIT.setBulkSourceLocations(c, getCluster().getFileSystem(),
+          getCluster().getTemporaryPath());
+    }
   }
 
   @ParameterizedTest

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

@@ -83,6 +83,7 @@ public void testBulkFile() throws Exception {
       Configuration conf = new Configuration();
       AccumuloConfiguration aconf = getCluster().getServerContext().getConfiguration();
       FileSystem fs = getCluster().getFileSystem();
+      BulkIT.setBulkSourceLocations(c, fs, getCluster().getTemporaryPath());
       String rootPath = cluster.getTemporaryPath().toString();
 
       String dir = rootPath + "/bulk_test_diff_files_89723987592_" + getUniqueNames(1)[0];
@@ -128,6 +129,7 @@ public void testExceptionInMetadataUpdate() throws Exception {
       Configuration conf = new Configuration();
       AccumuloConfiguration aconf = getCluster().getServerContext().getConfiguration();
       FileSystem fs = getCluster().getFileSystem();
+      BulkIT.setBulkSourceLocations(c, fs, getCluster().getTemporaryPath());
       String rootPath = cluster.getTemporaryPath().toString();
 
       String dir = rootPath + "/bulk_test_diff_files_89723987592_" + getUniqueNames(1)[0];

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

@@ -95,6 +95,7 @@ public void testBulkSplitOptimization() throws Exception {
       FileSystem fs = cluster.getFileSystem();
       Path testDir = new Path(cluster.getTemporaryPath(), "testmf");
       fs.deleteOnExit(testDir);
+      BulkIT.setBulkSourceLocations(c, fs, cluster.getTemporaryPath());
       FunctionalTestUtils.createRFiles(c, fs, testDir.toString(), ROWS, SPLITS, 8);
       FileStatus[] stats = fs.listStatus(testDir);
 

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

@@ -626,6 +626,7 @@ public void testSuccessfulCompaction() throws Exception {
       fs.deleteOnExit(testrf);
       FunctionalTestUtils.createRFiles(c, fs, testrf.toString(), 500000, 59, 4);
 
+      BulkIT.setBulkSourceLocations(c, fs, root);
       c.tableOperations().importDirectory(testrf.toString()).to(tableName).load();
       int beforeCount = countFiles(c, tableName);
 

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

@@ -247,6 +247,7 @@ public void bulkImportThatCantSplitHangsCompaction() throws Exception {
       Random random = new Random();
       byte[] val = new byte[100];
 
+      BulkIT.setBulkSourceLocations(c, getFileSystem(), getCluster().getTemporaryPath());
       String dir = getDir();
       String file = dir + "/f1.rf";