HADOOP-14524. Make CryptoCodec Closeable so it can be cleaned up proactively.

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/AesCtrCryptoCodec.java

@@ -22,6 +22,8 @@ import org.apache.hadoop.classification.InterfaceStability;
 
 import com.google.common.base.Preconditions;
 
+import java.io.IOException;
+
 @InterfaceAudience.Private
 @InterfaceStability.Evolving
 public abstract class AesCtrCryptoCodec extends CryptoCodec {
@@ -61,4 +63,8 @@ public abstract class AesCtrCryptoCodec extends CryptoCodec {
       IV[i] = (byte) sum;
     }
   }
+
+  @Override
+  public void close() throws IOException {
+  }
 }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/CryptoCodec.java

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.crypto;
 
+import java.io.Closeable;
 import java.security.GeneralSecurityException;
 import java.util.List;
 
@@ -42,7 +43,7 @@ import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_SECURITY
  */
 @InterfaceAudience.Private
 @InterfaceStability.Evolving
-public abstract class CryptoCodec implements Configurable {
+public abstract class CryptoCodec implements Configurable, Closeable {
   public static Logger LOG = LoggerFactory.getLogger(CryptoCodec.class);
   
   /**

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/CryptoInputStream.java

@@ -315,6 +315,7 @@ public class CryptoInputStream extends FilterInputStream implements
     
     super.close();
     freeBuffers();
+    codec.close();
     closed = true;
   }
   

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/CryptoOutputStream.java

@@ -239,6 +239,7 @@ public class CryptoOutputStream extends FilterOutputStream implements
       flush();
       if (closeOutputStream) {
         super.close();
+        codec.close();
       }
       freeBuffers();
     } finally {

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/OpensslAesCtrCryptoCodec.java

@@ -19,6 +19,7 @@ package org.apache.hadoop.crypto;
 
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_SECURITY_SECURE_RANDOM_IMPL_KEY;
 
+import java.io.Closeable;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.security.GeneralSecurityException;
@@ -90,6 +91,16 @@ public class OpensslAesCtrCryptoCodec extends AesCtrCryptoCodec {
     random.nextBytes(bytes);
   }
 
+  @Override
+  public void close() throws IOException {
+    try {
+      Closeable r = (Closeable) this.random;
+      r.close();
+    } catch (ClassCastException e) {
+    }
+    super.close();
+  }
+
   private static class OpensslAesCtrCipher implements Encryptor, Decryptor {
     private final OpensslCipher cipher;
     private final int mode;

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/key/KeyProviderCryptoExtension.java

@@ -254,6 +254,7 @@ public class KeyProviderCryptoExtension extends
       // Generate random bytes for new key and IV
 
       CryptoCodec cc = CryptoCodec.getInstance(keyProvider.getConf());
+      try {
         final byte[] newKey = new byte[encryptionKey.getMaterial().length];
         cc.generateSecureRandom(newKey);
         final byte[] iv = new byte[cc.getCipherSuite().getAlgorithmBlockSize()];
@@ -274,6 +275,9 @@ public class KeyProviderCryptoExtension extends
         return new EncryptedKeyVersion(encryptionKeyName,
             encryptionKey.getVersionName(), iv,
             new KeyVersion(encryptionKey.getName(), EEK, encryptedKey));
+      } finally {
+        cc.close();
+      }
     }
 
     @Override
@@ -300,6 +304,7 @@ public class KeyProviderCryptoExtension extends
           EncryptedKeyVersion.deriveIV(encryptedKeyVersion.getEncryptedKeyIv());
 
       CryptoCodec cc = CryptoCodec.getInstance(keyProvider.getConf());
+      try {
         Decryptor decryptor = cc.createDecryptor();
         decryptor.init(encryptionKey.getMaterial(), encryptionIV);
         final KeyVersion encryptedKV =
@@ -314,6 +319,9 @@ public class KeyProviderCryptoExtension extends
         byte[] decryptedKey = new byte[keyLen];
         bbOut.get(decryptedKey);
         return new KeyVersion(encryptionKey.getName(), EK, decryptedKey);
+      } finally {
+        cc.close();
+      }
     }
 
     @Override

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/crypto/TestCryptoStreamsWithOpensslAesCtrCryptoCodec.java

@@ -18,12 +18,17 @@
 package org.apache.hadoop.crypto;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.crypto.random.OsSecureRandom;
 import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
 import org.apache.hadoop.test.GenericTestUtils;
 import org.junit.BeforeClass;
+import org.junit.Test;
+import org.mockito.internal.util.reflection.Whitebox;
 
+import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_SECURITY_CRYPTO_CODEC_CLASSES_AES_CTR_NOPADDING_KEY;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 
 public class TestCryptoStreamsWithOpensslAesCtrCryptoCodec 
     extends TestCryptoStreams {
@@ -32,8 +37,7 @@ public class TestCryptoStreamsWithOpensslAesCtrCryptoCodec
   public static void init() throws Exception {
     GenericTestUtils.assumeInNativeProfile();
     Configuration conf = new Configuration();
-    conf.set(
+    conf.set(HADOOP_SECURITY_CRYPTO_CODEC_CLASSES_AES_CTR_NOPADDING_KEY,
-        CommonConfigurationKeysPublic.HADOOP_SECURITY_CRYPTO_CODEC_CLASSES_AES_CTR_NOPADDING_KEY,
         OpensslAesCtrCryptoCodec.class.getName());
     codec = CryptoCodec.getInstance(conf);
     assertNotNull("Unable to instantiate codec " +
@@ -42,4 +46,28 @@ public class TestCryptoStreamsWithOpensslAesCtrCryptoCodec
     assertEquals(OpensslAesCtrCryptoCodec.class.getCanonicalName(),
         codec.getClass().getCanonicalName());
   }
+
+  @Test
+  public void testCodecClosesRandom() throws Exception {
+    GenericTestUtils.assumeInNativeProfile();
+    Configuration conf = new Configuration();
+    conf.set(HADOOP_SECURITY_CRYPTO_CODEC_CLASSES_AES_CTR_NOPADDING_KEY,
+        OpensslAesCtrCryptoCodec.class.getName());
+    conf.set(
+        CommonConfigurationKeysPublic.HADOOP_SECURITY_SECURE_RANDOM_IMPL_KEY,
+        OsSecureRandom.class.getName());
+    CryptoCodec codecWithRandom = CryptoCodec.getInstance(conf);
+    assertNotNull(
+        "Unable to instantiate codec " + OpensslAesCtrCryptoCodec.class
+            .getName() + ", is the required " + "version of OpenSSL installed?",
+        codecWithRandom);
+    OsSecureRandom random =
+        (OsSecureRandom) Whitebox.getInternalState(codecWithRandom, "random");
+    // trigger the OsSecureRandom to create an internal FileInputStream
+    random.nextBytes(new byte[10]);
+    assertNotNull(Whitebox.getInternalState(random, "stream"));
+    // verify closing the codec closes the codec's random's stream.
+    codecWithRandom.close();
+    assertNull(Whitebox.getInternalState(random, "stream"));
+  }
 }

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/sasl/DataTransferSaslUtil.java

@@ -286,6 +286,7 @@ public final class DataTransferSaslUtil {
           codec.generateSecureRandom(inIv);
           codec.generateSecureRandom(outKey);
           codec.generateSecureRandom(outIv);
+          codec.close();
           return new CipherOption(suite, inKey, inIv, outKey, outIv);
         }
       }

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/CryptoUtils.java

@@ -66,16 +66,24 @@ public class CryptoUtils {
     if (isEncryptedSpillEnabled(conf)) {
       byte[] iv = new byte[cryptoCodec.getCipherSuite().getAlgorithmBlockSize()];
       cryptoCodec.generateSecureRandom(iv);
+      cryptoCodec.close();
       return iv;
     } else {
       return null;
     }
   }
 
-  public static int cryptoPadding(Configuration conf) {
+  public static int cryptoPadding(Configuration conf) throws IOException {
     // Sizeof(IV) + long(start-offset)
-    return isEncryptedSpillEnabled(conf) ? CryptoCodec.getInstance(conf)
+    if (!isEncryptedSpillEnabled(conf)) {
-        .getCipherSuite().getAlgorithmBlockSize() + 8 : 0;
+      return 0;
+    }
+    final CryptoCodec cryptoCodec = CryptoCodec.getInstance(conf);
+    try {
+      return cryptoCodec.getCipherSuite().getAlgorithmBlockSize() + 8;
+    } finally {
+      cryptoCodec.close();
+    }
   }
 
   private static byte[] getEncryptionKey() throws IOException {