Merge branch '3.1' into zkchroot-postzkclean

Author: ctubbsii

core/src/main/java/org/apache/accumulo/core/spi/crypto/AESCryptoService.java

@@ -426,20 +426,21 @@ public byte[] getDecryptionParameters() {
     }
 
     public class AESGCMFileDecrypter implements FileDecrypter {
-      private final Cipher cipher;
       private final Key fek;
 
       AESGCMFileDecrypter(Key fek) {
+        this.fek = fek;
+      }
+
+      @Override
+      public InputStream decryptStream(InputStream inputStream) throws CryptoException {
+        Cipher cipher;
         try {
           cipher = Cipher.getInstance(transformation);
         } catch (NoSuchAlgorithmException | NoSuchPaddingException e) {
           throw new CryptoException("Error obtaining cipher for transform " + transformation, e);
         }
-        this.fek = fek;
-      }
 
-      @Override
-      public InputStream decryptStream(InputStream inputStream) throws CryptoException {
         byte[] initVector = new byte[GCM_IV_LENGTH_IN_BYTES];
         try {
           IOUtils.readFully(inputStream, initVector);
@@ -531,20 +532,21 @@ public byte[] getDecryptionParameters() {
 
     @SuppressFBWarnings(value = "CIPHER_INTEGRITY", justification = "CBC is provided for WALs")
     public class AESCBCFileDecrypter implements FileDecrypter {
-      private final Cipher cipher;
       private final Key fek;
 
       AESCBCFileDecrypter(Key fek) {
+        this.fek = fek;
+      }
+
+      @Override
+      public InputStream decryptStream(InputStream inputStream) throws CryptoException {
+        Cipher cipher;
         try {
           cipher = Cipher.getInstance(transformation);
         } catch (NoSuchAlgorithmException | NoSuchPaddingException e) {
           throw new CryptoException("Error obtaining cipher for transform " + transformation, e);
         }
-        this.fek = fek;
-      }
 
-      @Override
-      public InputStream decryptStream(InputStream inputStream) throws CryptoException {
         byte[] initVector = new byte[IV_LENGTH_IN_BYTES];
         try {
           IOUtils.readFully(inputStream, initVector);

core/src/test/java/org/apache/accumulo/core/crypto/CryptoTest.java

@@ -30,6 +30,7 @@
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
 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 java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
@@ -46,7 +47,10 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 
 import javax.crypto.Cipher;
 import javax.crypto.NoSuchPaddingException;
@@ -497,6 +501,67 @@ public void testPerTableFactory() {
     assertEquals(2, factory.getCount());
   }
 
+  @Test
+  public void testMultipleThreads() throws Exception {
+    testMultipleThreads(WAL);
+    testMultipleThreads(TABLE);
+  }
+
+  private void testMultipleThreads(Scope scope) throws Exception {
+
+    byte[] plainText = new byte[1024 * 1024];
+    for (int i = 0; i < plainText.length; i++) {
+      plainText[i] = (byte) (i % 128);
+    }
+
+    AESCryptoService cs = new AESCryptoService();
+    cs.init(getAllCryptoProperties(ConfigMode.CRYPTO_TABLE_ON));
+    CryptoEnvironment encEnv = new CryptoEnvironmentImpl(scope, null, null);
+    FileEncrypter encrypter = cs.getFileEncrypter(encEnv);
+    byte[] params = encrypter.getDecryptionParameters();
+
+    assertNotNull(params);
+
+    ByteArrayOutputStream out = new ByteArrayOutputStream();
+    DataOutputStream dataOut = new DataOutputStream(out);
+    OutputStream encrypted = encrypter.encryptStream(dataOut);
+
+    assertNotNull(encrypted);
+    DataOutputStream cipherOut = new DataOutputStream(encrypted);
+
+    cipherOut.write(plainText);
+
+    cipherOut.close();
+    dataOut.close();
+    encrypted.close();
+    out.close();
+    byte[] cipherText = out.toByteArray();
+
+    var executor = Executors.newCachedThreadPool();
+
+    List<Future<Boolean>> verifyFutures = new ArrayList<>();
+
+    FileDecrypter decrypter = cs.getFileDecrypter(new CryptoEnvironmentImpl(scope, null, params));
+
+    // verify that each input stream returned by decrypter.decryptStream() is independent when used
+    // by multiple threads
+    for (int i = 0; i < 32; i++) {
+      var future = executor.submit(() -> {
+        try (ByteArrayInputStream in = new ByteArrayInputStream(cipherText);
+            DataInputStream decrypted = new DataInputStream(decrypter.decryptStream(in))) {
+          byte[] dataRead = new byte[plainText.length];
+          decrypted.readFully(dataRead);
+          return Arrays.equals(plainText, dataRead);
+        }
+      });
+      verifyFutures.add(future);
+    }
+
+    for (var future : verifyFutures) {
+      assertTrue(future.get());
+    }
+  }
+
   private ArrayList<Key> testData() {
     ArrayList<Key> keys = new ArrayList<>();
     keys.add(new Key("a", "cf", "cq"));