Implement new Encryption interface WIP

core/src/main/java/org/apache/accumulo/core/client/rfile/RFileScanner.java

@@ -61,6 +61,8 @@
 import org.apache.accumulo.core.iterators.system.MultiIterator;
 import org.apache.accumulo.core.sample.impl.SamplerConfigurationImpl;
 import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.security.crypto.CryptoService;
+import org.apache.accumulo.core.security.crypto.CryptoServiceFactory;
 import org.apache.accumulo.core.util.LocalityGroupUtil;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.io.Text;
@@ -80,6 +82,7 @@ class RFileScanner extends ScannerOptions implements Scanner {
   private int batchSize = 1000;
   private long readaheadThreshold = 3;
   private AccumuloConfiguration tableConf;
+  private CryptoService cryptoService;
 
   static class Opts {
     InputArgs in;
@@ -215,6 +218,7 @@ public void indexWeightChanged() {}
     if (null == this.dataCache) {
       this.dataCache = new NoopCache();
     }
+    this.cryptoService = CryptoServiceFactory.getConfigured(tableConf);
   }
 
   @Override
@@ -353,8 +357,9 @@ public Iterator<Entry<Key,Value>> iterator() {
       for (int i = 0; i < sources.length; i++) {
         // TODO may have been a bug with multiple files and caching in older version...
         FSDataInputStream inputStream = (FSDataInputStream) sources[i].getInputStream();
-        readers.add(new RFile.Reader(new CachableBlockFile.Reader("source-" + i, inputStream,
-            sources[i].getLength(), opts.in.getConf(), dataCache, indexCache, tableConf)));
+        readers.add(new RFile.Reader(
+            new CachableBlockFile.Reader("source-" + i, inputStream, sources[i].getLength(),
+                opts.in.getConf(), dataCache, indexCache, tableConf, cryptoService)));
       }
 
       if (getSamplerConfiguration() != null) {

core/src/main/java/org/apache/accumulo/core/client/rfile/RFileSummariesRetriever.java

@@ -31,6 +31,7 @@
 import org.apache.accumulo.core.client.summary.Summary;
 import org.apache.accumulo.core.conf.AccumuloConfiguration;
 import org.apache.accumulo.core.conf.DefaultConfiguration;
+import org.apache.accumulo.core.security.crypto.CryptoServiceFactory;
 import org.apache.accumulo.core.summary.Gatherer;
 import org.apache.accumulo.core.summary.SummarizerFactory;
 import org.apache.accumulo.core.summary.SummaryCollection;
@@ -88,7 +89,8 @@ public Collection<Summary> read() throws IOException {
       SummaryCollection all = new SummaryCollection();
       for (RFileSource source : sources) {
         SummaryReader fileSummary = SummaryReader.load(conf, acuconf, source.getInputStream(),
-            source.getLength(), summarySelector, factory);
+            source.getLength(), summarySelector, factory,
+            CryptoServiceFactory.getConfigured(acuconf));
         SummaryCollection sc = fileSummary
             .getSummaries(Collections.singletonList(new Gatherer.RowRange(startRow, endRow)));
         all.merge(sc, factory);

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

@@ -20,8 +20,7 @@
 import java.util.Map.Entry;
 import java.util.Objects;
 
-import org.apache.accumulo.core.security.crypto.CryptoModule;
-import org.apache.accumulo.core.security.crypto.SecretKeyEncryptionStrategy;
+import org.apache.accumulo.core.security.crypto.CryptoService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -34,9 +33,6 @@ public class ConfigSanityCheck {
 
   private static final Logger log = LoggerFactory.getLogger(ConfigSanityCheck.class);
   private static final String PREFIX = "BAD CONFIG ";
-  private static final String NULL_CIPHER = "NullCipher";
-  private static final String NULL_CRYPTO_MODULE = "NullCryptoModule";
-  private static final String NULL_SECRET_KEY_CRYPT_STRATEGY = "NullSecretKeyEncryptionStrategy";
   @SuppressWarnings("deprecation")
   private static final Property INSTANCE_DFS_URI = Property.INSTANCE_DFS_URI;
   @SuppressWarnings("deprecation")
@@ -56,10 +52,6 @@ public class ConfigSanityCheck {
   public static void validate(Iterable<Entry<String,String>> entries) {
     String instanceZkTimeoutValue = null;
     boolean usingVolumes = false;
-    String cipherSuite = NULL_CIPHER;
-    String keyAlgorithm = NULL_CIPHER;
-    String secretKeyEncryptionStrategy = NULL_SECRET_KEY_CRYPT_STRATEGY;
-    String cryptoModule = NULL_CRYPTO_MODULE;
     for (Entry<String,String> entry : entries) {
       String key = entry.getKey();
       String value = entry.getValue();
@@ -91,30 +83,9 @@ else if (!prop.getType().isValidFormat(value))
             + " must be greater than 0 and less than " + Integer.MAX_VALUE + " but was: " + bsize);
       }
 
-      if (key.equals(Property.CRYPTO_CIPHER_SUITE.getKey())) {
-        cipherSuite = Objects.requireNonNull(value);
-        Preconditions.checkArgument(
-            cipherSuite.equals(NULL_CIPHER) || cipherSuite.split("/").length == 3,
-            "Cipher suite must be NullCipher or in the form algorithm/mode/padding. Suite: "
-                + cipherSuite + " is invalid.");
-      }
-
-      if (key.equals(Property.CRYPTO_CIPHER_KEY_ALGORITHM_NAME.getKey())) {
-        keyAlgorithm = Objects.requireNonNull(value);
-      }
-
-      if (key.equals(Property.CRYPTO_MODULE_CLASS.getKey())) {
-        cryptoModule = Objects.requireNonNull(value);
-        if (!cryptoModule.equals(NULL_CRYPTO_MODULE)) {
-          verifyValidClassName(key, cryptoModule, CryptoModule.class);
-        }
-
-      }
-      if (key.equals(Property.CRYPTO_SECRET_KEY_ENCRYPTION_STRATEGY_CLASS.getKey())) {
-        secretKeyEncryptionStrategy = Objects.requireNonNull(value);
-        if (!secretKeyEncryptionStrategy.equals(NULL_SECRET_KEY_CRYPT_STRATEGY)) {
-          verifyValidClassName(key, secretKeyEncryptionStrategy, SecretKeyEncryptionStrategy.class);
-        }
+      if (key.equals(Property.TABLE_CRYPTO_SERVICE.getKey())) {
+        String cryptoStrategy = Objects.requireNonNull(value);
+        verifyValidClassName(key, cryptoStrategy, CryptoService.class);
       }
     }
 
@@ -127,19 +98,6 @@ else if (!prop.getType().isValidFormat(value))
       log.warn("Use of {} and {} are deprecated. Consider using {} instead.", INSTANCE_DFS_URI,
           INSTANCE_DFS_DIR, Property.INSTANCE_VOLUMES);
     }
-
-    if ((cipherSuite.equals(NULL_CIPHER) || keyAlgorithm.equals(NULL_CIPHER))
-        && !cipherSuite.equals(keyAlgorithm)) {
-      fatal(Property.CRYPTO_CIPHER_SUITE.getKey() + " and "
-          + Property.CRYPTO_CIPHER_KEY_ALGORITHM_NAME + " must both be configured.");
-    }
-
-    if (cryptoModule.equals(NULL_CRYPTO_MODULE)
-        ^ secretKeyEncryptionStrategy.equals(NULL_SECRET_KEY_CRYPT_STRATEGY)) {
-      fatal(Property.CRYPTO_MODULE_CLASS.getKey() + " and "
-          + Property.CRYPTO_SECRET_KEY_ENCRYPTION_STRATEGY_CLASS.getKey()
-          + " must both be configured.");
-    }
   }
 
   private interface CheckTimeDuration {

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

@@ -42,76 +42,6 @@
 import com.google.common.base.Preconditions;
 
 public enum Property {
-  // Crypto-related properties
-  @Experimental
-  CRYPTO_PREFIX("crypto.", null, PropertyType.PREFIX,
-      "Properties in this category related to the configuration of both default and custom crypto"
-          + " modules."),
-  @Experimental
-  CRYPTO_MODULE_CLASS("crypto.module.class", "NullCryptoModule", PropertyType.STRING,
-      "Fully qualified class name of the class that implements the CryptoModule"
-          + " interface, to be used in setting up encryption at rest for the WAL and"
-          + " (future) other parts of the code."),
-  @Experimental
-  CRYPTO_CIPHER_SUITE("crypto.cipher.suite", "NullCipher", PropertyType.STRING,
-      "Describes the cipher suite to use for rfile encryption. The value must"
-          + " be either NullCipher or in the form of algorithm/mode/padding, e.g."
-          + " AES/CBC/NoPadding"),
-  @Experimental
-  CRYPTO_WAL_CIPHER_SUITE("crypto.wal.cipher.suite", "", PropertyType.STRING,
-      "Describes the cipher suite to use for the write-ahead log. Defaults to"
-          + " 'cyrpto.cipher.suite' and will use that value for WAL encryption unless"
-          + " otherwise specified. Valid suite values include: an empty string,"
-          + " NullCipher, or a string the form of algorithm/mode/padding, e.g."
-          + " AES/CBC/NOPadding"),
-  @Experimental
-  CRYPTO_CIPHER_KEY_ALGORITHM_NAME("crypto.cipher.key.algorithm.name", "NullCipher",
-      PropertyType.STRING,
-      "States the name of the algorithm used for the key for the corresponding"
-          + " cipher suite. The key type must be compatible with the cipher suite."),
-  @Experimental
-  CRYPTO_BLOCK_STREAM_SIZE("crypto.block.stream.size", "1K", PropertyType.BYTES,
-      "The size of the buffer above the cipher stream. Used for reading files"
-          + " and padding walog entries."),
-  @Experimental
-  CRYPTO_CIPHER_KEY_LENGTH("crypto.cipher.key.length", "128", PropertyType.STRING,
-      "Specifies the key length *in bits* to use for the symmetric key, "
-          + "should probably be 128 or 256 unless you really know what you're doing"),
-  @Experimental
-  CRYPTO_SECURITY_PROVIDER("crypto.security.provider", "", PropertyType.STRING,
-      "States the security provider to use, and defaults to the system configured provider"),
-  @Experimental
-  CRYPTO_SECURE_RNG("crypto.secure.rng", "SHA1PRNG", PropertyType.STRING,
-      "States the secure random number generator to use, and defaults to the built-in SHA1PRNG"),
-  @Experimental
-  CRYPTO_SECURE_RNG_PROVIDER("crypto.secure.rng.provider", "SUN", PropertyType.STRING,
-      "States the secure random number generator provider to use."),
-  @Experimental
-  CRYPTO_SECRET_KEY_ENCRYPTION_STRATEGY_CLASS("crypto.secret.key.encryption.strategy.class",
-      "NullSecretKeyEncryptionStrategy", PropertyType.STRING,
-      "The class Accumulo should use for its key encryption strategy."),
-  @Experimental
-  CRYPTO_DEFAULT_KEY_STRATEGY_KEY_LOCATION("crypto.default.key.strategy.key.location",
-      "/crypto/secret/keyEncryptionKey", PropertyType.ABSOLUTEPATH,
-      "The path relative to the top level instance directory (instance.dfs.dir) where to store"
-          + " the key encryption key within HDFS."),
-  @Experimental
-  CRYPTO_DEFAULT_KEY_STRATEGY_CIPHER_SUITE("crypto.default.key.strategy.cipher.suite", "NullCipher",
-      PropertyType.STRING,
-      "The cipher suite to use when encrypting session keys with a key"
-          + " encryption keyThis should be set to match the overall encryption"
-          + " algorithm but with ECB mode and no padding unless you really know what"
-          + " you're doing and are sure you won't break internal file formats"),
-  @Experimental
-  CRYPTO_OVERRIDE_KEY_STRATEGY_WITH_CONFIGURED_STRATEGY(
-      "crypto.override.key.strategy.with.configured.strategy", "false", PropertyType.BOOLEAN,
-      "The default behavior is to record the key encryption strategy with the"
-          + " encrypted file, and continue to use that strategy for the life of that"
-          + " file. Sometimes, you change your strategy and want to use the new"
-          + " strategy, not the old one. (Most commonly, this will be because you have"
-          + " moved key material from one spot to another.) If you want to override"
-          + " the recorded key strategy with the one in the configuration file, set"
-          + " this property to true."),
   // SSL properties local to each node (see also instance.ssl.enabled which must be consistent
   // across all nodes in an instance)
   RPC_PREFIX("rpc.", null, PropertyType.PREFIX,
@@ -833,6 +763,20 @@ public enum Property {
           + " To add a summarizer set table.summarizer.<unique id>=<summarizer class"
           + " name>. If the summarizer has options, then for each option set"
           + " table.summarizer.<unique id>.opt.<key>=<value>."),
+  // Crypto-related properties
+  @Experimental
+  TABLE_CRYPTO_PREFIX("table.crypto.opts.", null, PropertyType.PREFIX,
+      "Properties related to on-disk file encryption."),
+  @Experimental
+  @Sensitive
+  TABLE_CRYPTO_SENSITIVE_PREFIX("table.crypto.opts.sensitive.", null, PropertyType.PREFIX,
+      "Sensitive properties related to on-disk file encryption."),
+  @Experimental
+  TABLE_CRYPTO_SERVICE("table.crypto.service",
+      "org.apache.accumulo.core.security.crypto.impl.NoCryptoService", PropertyType.CLASSNAME,
+      "The class which executes on-disk file encryption. The default does nothing. To enable "
+          + "encryption, replace this classname with an implementation of the"
+          + "org.apache.accumulo.core.security.crypto.CryptoService interface."),
 
   // VFS ClassLoader properties
   VFS_CLASSLOADER_SYSTEM_CLASSPATH_PROPERTY(
@@ -1151,6 +1095,7 @@ public static boolean isValidTablePropertyKey(String key) {
             || key.startsWith(Property.TABLE_REPLICATION_TARGET.getKey())
             || key.startsWith(Property.TABLE_ARBITRARY_PROP_PREFIX.getKey())
             || key.startsWith(TABLE_SAMPLER_OPTS.getKey())
+            || key.startsWith(TABLE_CRYPTO_PREFIX.getKey())
             || key.startsWith(TABLE_SUMMARIZER_PREFIX.getKey())));
   }
 

core/src/main/java/org/apache/accumulo/core/file/FileOperations.java

@@ -29,6 +29,7 @@
 import org.apache.accumulo.core.data.Range;
 import org.apache.accumulo.core.file.blockfile.cache.BlockCache;
 import org.apache.accumulo.core.file.rfile.RFile;
+import org.apache.accumulo.core.security.crypto.CryptoService;
 import org.apache.accumulo.core.util.ratelimit.RateLimiter;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FSDataOutputStream;
@@ -395,6 +396,7 @@ protected static class FileReaderOperation<SubclassType extends FileReaderOperat
     private BlockCache dataCache;
     private BlockCache indexCache;
     private Cache<String,Long> fileLenCache;
+    private CryptoService cryptoService;
 
     /**
      * (Optional) Set the block cache pair to be used to optimize reads within the constructed
@@ -429,6 +431,11 @@ public SubclassType withFileLenCache(Cache<String,Long> fileLenCache) {
       return (SubclassType) this;
     }
 
+    public SubclassType withCryptoService(CryptoService cryptoService) {
+      this.cryptoService = cryptoService;
+      return (SubclassType) this;
+    }
+
     public BlockCache getDataCache() {
       return dataCache;
     }
@@ -440,6 +447,10 @@ public BlockCache getIndexCache() {
     public Cache<String,Long> getFileLenCache() {
       return fileLenCache;
     }
+
+    public CryptoService getCryptoService() {
+      return cryptoService;
+    }
   }
 
   /** Builder interface parallel to {@link FileReaderOperation}. */

core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/CachableBlockFile.java

@@ -36,6 +36,7 @@
 import org.apache.accumulo.core.file.rfile.bcfile.BCFile.Reader.BlockReader;
 import org.apache.accumulo.core.file.rfile.bcfile.MetaBlockDoesNotExist;
 import org.apache.accumulo.core.file.streams.RateLimitedInputStream;
+import org.apache.accumulo.core.security.crypto.CryptoService;
 import org.apache.accumulo.core.util.ratelimit.RateLimiter;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
@@ -75,6 +76,7 @@ public static class Reader implements Closeable {
     private boolean closed = false;
     private final Configuration conf;
     private final AccumuloConfiguration accumuloConfiguration;
+    private final CryptoService cryptoService;
 
     private final IoeSupplier<InputStream> inputSupplier;
     private final IoeSupplier<Long> lengthSupplier;
@@ -105,23 +107,25 @@ private BCFile.Reader getBCFile(byte[] serializedMetadata) throws IOException {
         BCFile.Reader tmpReader = null;
         if (serializedMetadata == null) {
           if (fileLenCache == null) {
-            tmpReader = new BCFile.Reader(fsIn, lengthSupplier.get(), conf, accumuloConfiguration);
+            tmpReader = new BCFile.Reader(fsIn, lengthSupplier.get(), conf, accumuloConfiguration,
+                cryptoService);
           } else {
             long len = getCachedFileLen();
             try {
-              tmpReader = new BCFile.Reader(fsIn, len, conf, accumuloConfiguration);
+              tmpReader = new BCFile.Reader(fsIn, len, conf, accumuloConfiguration, cryptoService);
             } catch (Exception e) {
               log.debug("Failed to open {}, clearing file length cache and retrying", cacheId, e);
               fileLenCache.invalidate(cacheId);
             }
 
             if (tmpReader == null) {
               len = getCachedFileLen();
-              tmpReader = new BCFile.Reader(fsIn, len, conf, accumuloConfiguration);
+              tmpReader = new BCFile.Reader(fsIn, len, conf, accumuloConfiguration, cryptoService);
             }
           }
         } else {
-          tmpReader = new BCFile.Reader(serializedMetadata, fsIn, conf);
+          tmpReader = new BCFile.Reader(serializedMetadata, fsIn, conf, accumuloConfiguration,
+              cryptoService);
         }
 
         if (!bcfr.compareAndSet(null, tmpReader)) {
@@ -297,7 +301,7 @@ public byte[] load(int maxSize, Map<String,byte[]> dependencies) {
     private Reader(String cacheId, IoeSupplier<InputStream> inputSupplier,
         IoeSupplier<Long> lenghtSupplier, Cache<String,Long> fileLenCache, BlockCache data,
         BlockCache index, RateLimiter readLimiter, Configuration conf,
-        AccumuloConfiguration accumuloConfiguration) {
+        AccumuloConfiguration accumuloConfiguration, CryptoService cryptoService) {
       Preconditions.checkArgument(cacheId != null || (data == null && index == null));
       this.cacheId = cacheId;
       this.inputSupplier = inputSupplier;
@@ -308,29 +312,36 @@ private Reader(String cacheId, IoeSupplier<InputStream> inputSupplier,
       this.readLimiter = readLimiter;
       this.conf = conf;
       this.accumuloConfiguration = accumuloConfiguration;
+      this.cryptoService = cryptoService;
     }
 
     public Reader(FileSystem fs, Path dataFile, Configuration conf, BlockCache data,
-        BlockCache index, AccumuloConfiguration accumuloConfiguration) throws IOException {
-      this(fs, dataFile, conf, null, data, index, null, accumuloConfiguration);
+        BlockCache index, AccumuloConfiguration accumuloConfiguration, CryptoService cryptoService)
+        throws IOException {
+      this(fs, dataFile, conf, null, data, index, null, accumuloConfiguration, cryptoService);
     }
 
     public Reader(FileSystem fs, Path dataFile, Configuration conf, Cache<String,Long> fileLenCache,
         BlockCache data, BlockCache index, RateLimiter readLimiter,
-        AccumuloConfiguration accumuloConfiguration) throws IOException {
+        AccumuloConfiguration accumuloConfiguration, CryptoService cryptoService)
+        throws IOException {
       this(dataFile.toString(), () -> fs.open(dataFile), () -> fs.getFileStatus(dataFile).getLen(),
-          fileLenCache, data, index, readLimiter, conf, accumuloConfiguration);
+          fileLenCache, data, index, readLimiter, conf, accumuloConfiguration, cryptoService);
     }
 
     public <InputStreamType extends InputStream & Seekable> Reader(String cacheId,
         InputStreamType fsin, long len, Configuration conf, BlockCache data, BlockCache index,
-        AccumuloConfiguration accumuloConfiguration) throws IOException {
-      this(cacheId, () -> fsin, () -> len, null, data, index, null, conf, accumuloConfiguration);
+        AccumuloConfiguration accumuloConfiguration, CryptoService cryptoService)
+        throws IOException {
+      this(cacheId, () -> fsin, () -> len, null, data, index, null, conf, accumuloConfiguration,
+          cryptoService);
     }
 
     public <InputStreamType extends InputStream & Seekable> Reader(InputStreamType fsin, long len,
-        Configuration conf, AccumuloConfiguration accumuloConfiguration) throws IOException {
-      this(null, () -> fsin, () -> len, null, null, null, null, conf, accumuloConfiguration);
+        Configuration conf, AccumuloConfiguration accumuloConfiguration,
+        CryptoService cryptoService) throws IOException {
+      this(null, () -> fsin, () -> len, null, null, null, null, conf, accumuloConfiguration,
+          cryptoService);
     }
 
     /**