[2/2] FLUME-1424. File Channel should support encryption.

[mp...@apache.org]

http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/encryption/EncryptionTestUtils.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/encryption/EncryptionTestUtils.java b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/encryption/EncryptionTestUtils.java
new file mode 100644
index 0000000..a157661
--- /dev/null
+++ b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/encryption/EncryptionTestUtils.java
@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.flume.channel.file.encryption;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.security.Key;
+import java.security.KeyStore;
+import java.util.List;
+import java.util.Map;
+
+import javax.crypto.KeyGenerator;
+
+import org.apache.flume.channel.file.TestUtils;
+
+import com.google.common.base.Charsets;
+import com.google.common.base.Joiner;
+import com.google.common.base.Throwables;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.io.Files;
+import com.google.common.io.Resources;
+
+public class EncryptionTestUtils {
+
+  private static Key newKey() {
+    KeyGenerator keyGen;
+    try {
+      keyGen = KeyGenerator.getInstance("AES");
+      Key key = keyGen.generateKey();
+      return key;
+    } catch (Exception e) {
+      throw Throwables.propagate(e);
+    }
+  }
+  public static void createKeyStore(File keyStoreFile,
+      File keyStorePasswordFile, Map<String, File> keyAliasPassword)
+          throws Exception {
+    KeyStore ks = KeyStore.getInstance("jceks");
+    ks.load(null);
+    List<String> keysWithSeperatePasswords = Lists.newArrayList();
+    for(String alias : keyAliasPassword.keySet()) {
+      Key key = newKey();
+      char[] password = null;
+      File passwordFile = keyAliasPassword.get(alias);
+      if(passwordFile == null) {
+        password = Files.toString(keyStorePasswordFile, Charsets.UTF_8)
+            .toCharArray();
+      } else {
+        keysWithSeperatePasswords.add(alias);
+        password = Files.toString(passwordFile, Charsets.UTF_8).toCharArray();
+      }
+      ks.setKeyEntry(alias, key, password, null);
+    }
+    char[] keyStorePassword = Files.
+        toString(keyStorePasswordFile, Charsets.UTF_8).toCharArray();
+    FileOutputStream outputStream = new FileOutputStream(keyStoreFile);
+    ks.store(outputStream, keyStorePassword);
+    outputStream.close();
+  }
+  public static Map<String, File> configureTestKeyStore(File baseDir,
+      File keyStoreFile) throws IOException {
+    Map<String, File> result = Maps.newHashMap();
+    Resources.copy(Resources.getResource("test.keystore"),
+        new FileOutputStream(keyStoreFile));
+    /*
+    Commands below:
+    keytool -genseckey -alias key-0 -keypass keyPassword -keyalg AES \
+      -keysize 192 -validity 9000 -keystore src/test/resources/test.keystore \
+      -storetype jceks -storepass keyStorePassword
+    keytool -genseckey -alias key-1 -keyalg AES -keysize 192 -validity 9000 \
+      -keystore src/test/resources/test.keystore -storetype jceks \
+      -storepass keyStorePassword
+     */
+//  key-0 has own password, key-1 used key store password
+    result.put("key-0",
+        TestUtils.writeStringToFile(baseDir, "key-0", "keyPassword"));
+    result.put("key-1", null);
+    return result;
+  }
+  public static Map<String,String> configureForKeyStore(File keyStoreFile,
+      File keyStorePasswordFile, Map<String, File> keyAliasPassword)
+          throws Exception {
+    Map<String, String> context = Maps.newHashMap();
+    List<String> keys = Lists.newArrayList();
+    for(String alias : keyAliasPassword.keySet()) {
+      String propertyName = EncryptionConfiguration.KEYS + "." + alias + "." +
+          EncryptionConfiguration.JCE_FILE_KEY_PASSWORD_FILE;
+      File passwordFile = keyAliasPassword.get(alias);
+      if(passwordFile == null) {
+        keys.add(alias);
+        context.put(propertyName, keyStorePasswordFile.getAbsolutePath());
+      } else {
+        keys.add(alias);
+        context.put(propertyName, passwordFile.getAbsolutePath());
+      }
+    }
+    context.put(EncryptionConfiguration.JCE_FILE_KEY_STORE_FILE,
+        keyStoreFile.getAbsolutePath());
+    if(keyStorePasswordFile != null) {
+      context.put(EncryptionConfiguration.JCE_FILE_KEY_STORE_PASSWORD_FILE,
+          keyStorePasswordFile.getAbsolutePath());
+    }
+    context.put(EncryptionConfiguration.KEYS, Joiner.on(" ").join(keys));
+    return context;
+  }
+}

http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/encryption/TestAESCTRNoPaddingProvider.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/encryption/TestAESCTRNoPaddingProvider.java b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/encryption/TestAESCTRNoPaddingProvider.java
new file mode 100644
index 0000000..a7c7cb2
--- /dev/null
+++ b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/encryption/TestAESCTRNoPaddingProvider.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.flume.channel.file.encryption;
+
+import java.security.Key;
+
+import javax.crypto.KeyGenerator;
+
+import org.junit.Before;
+import org.junit.Test;
+
+public class TestAESCTRNoPaddingProvider {
+  private Key key;
+  private CipherProvider.Encryptor encryptor;
+  private CipherProvider.Decryptor decryptor;
+  private CipherProviderTestSuite cipherProviderTestSuite;
+
+  @Before
+  public void setup() throws Exception {
+    KeyGenerator keyGen = KeyGenerator.getInstance("AES");
+    key = keyGen.generateKey();
+    encryptor = CipherProviderFactory.
+        getEncrypter(CipherProviderType.AESCTRNOPADDING.name(), key);
+    decryptor = CipherProviderFactory.
+        getDecrypter(CipherProviderType.AESCTRNOPADDING.name(), key,
+            encryptor.getParameters());
+    cipherProviderTestSuite = new CipherProviderTestSuite(encryptor, decryptor);
+  }
+  @Test
+  public void test() throws Exception {
+    cipherProviderTestSuite.test();
+  }
+}

http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/encryption/TestFileChannelEncryption.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/encryption/TestFileChannelEncryption.java b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/encryption/TestFileChannelEncryption.java
new file mode 100644
index 0000000..7d7a825
--- /dev/null
+++ b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/encryption/TestFileChannelEncryption.java
@@ -0,0 +1,298 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.flume.channel.file.encryption;
+
+import static org.apache.flume.channel.file.TestUtils.*;
+
+import java.io.File;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.flume.ChannelException;
+import org.apache.flume.FlumeException;
+import org.apache.flume.channel.file.FileChannel;
+import org.apache.flume.channel.file.FileChannelConfiguration;
+import org.apache.flume.channel.file.TestUtils;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Charsets;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import com.google.common.io.Files;
+
+public class TestFileChannelEncryption {
+  protected static final Logger LOGGER =
+      LoggerFactory.getLogger(TestFileChannelEncryption.class);
+  private FileChannel channel;
+  private File baseDir;
+  private File checkpointDir;
+  private File[] dataDirs;
+  private String dataDir;
+  private File keyStoreFile;
+  private File keyStorePasswordFile;
+  private Map<String, File> keyAliasPassword;
+
+  @Before
+  public void setup() throws Exception {
+    baseDir = Files.createTempDir();
+    checkpointDir = new File(baseDir, "chkpt");
+    Assert.assertTrue(checkpointDir.mkdirs() || checkpointDir.isDirectory());
+    dataDirs = new File[3];
+    dataDir = "";
+    for (int i = 0; i < dataDirs.length; i++) {
+      dataDirs[i] = new File(baseDir, "data" + (i+1));
+      Assert.assertTrue(dataDirs[i].mkdirs() || dataDirs[i].isDirectory());
+      dataDir += dataDirs[i].getAbsolutePath() + ",";
+    }
+    dataDir = dataDir.substring(0, dataDir.length() - 1);
+    keyStorePasswordFile = new File(baseDir, "keyStorePasswordFile");
+    Files.write("keyStorePassword", keyStorePasswordFile, Charsets.UTF_8);
+    keyStoreFile = new File(baseDir, "keyStoreFile");
+    Assert.assertTrue(keyStoreFile.createNewFile());
+    keyAliasPassword = Maps.newHashMap();
+    keyAliasPassword.putAll(EncryptionTestUtils.
+        configureTestKeyStore(baseDir, keyStoreFile));
+  }
+  @After
+  public void teardown() {
+    if(channel != null && channel.isOpen()) {
+      channel.stop();
+    }
+    FileUtils.deleteQuietly(baseDir);
+  }
+  private Map<String, String> getOverrides() throws Exception {
+    Map<String, String> overrides = Maps.newHashMap();
+    overrides.put(FileChannelConfiguration.CAPACITY, String.valueOf(100));
+    return overrides;
+  }
+  private Map<String, String> getOverridesForEncryption() throws Exception {
+    Map<String, String> overrides = getOverrides();
+    Map<String, String> encryptionProps = EncryptionTestUtils.
+        configureForKeyStore(keyStoreFile, keyStorePasswordFile,
+            keyAliasPassword);
+    encryptionProps.put(EncryptionConfiguration.KEY_PROVIDER,
+        KeyProviderType.JCEKSFILE.name());
+    encryptionProps.put(EncryptionConfiguration.CIPHER_PROVIDER,
+        CipherProviderType.AESCTRNOPADDING.name());
+    encryptionProps.put(EncryptionConfiguration.KEY_ALIAS, "key-1");
+    for(String key : encryptionProps.keySet()) {
+      overrides.put(EncryptionConfiguration.ENCRYPTION_PREFIX + "." + key,
+          encryptionProps.get(key));
+    }
+    return overrides;
+  }
+  @Test
+  public void testBasicEncyrptionDecryption() throws Exception {
+    Map<String, String> overrides = getOverridesForEncryption();
+    channel = TestUtils.createFileChannel(checkpointDir.getAbsolutePath(),
+        dataDir, overrides);
+    channel.start();
+    Assert.assertTrue(channel.isOpen());
+    Set<String> in = Sets.newHashSet();
+    try {
+      while(true) {
+        in.addAll(putEvents(channel, "restart", 1, 1));
+      }
+    } catch (ChannelException e) {
+      Assert.assertEquals("Cannot acquire capacity. [channel="
+          +channel.getName()+"]", e.getMessage());
+    }
+    channel.stop();
+    channel = TestUtils.createFileChannel(checkpointDir.getAbsolutePath(),
+        dataDir, overrides);
+    channel.start();
+    Assert.assertTrue(channel.isOpen());
+    Set<String> out = takeEvents(channel, 1, Integer.MAX_VALUE);
+    compareInputAndOut(in, out);
+  }
+  @Test
+  public void testEncryptedChannelWithoutEncryptionConfigFails() throws Exception {
+    Map<String, String> overrides = getOverridesForEncryption();
+    channel = TestUtils.createFileChannel(checkpointDir.getAbsolutePath(),
+        dataDir, overrides);
+    channel.start();
+    Assert.assertTrue(channel.isOpen());
+    Set<String> in = Sets.newHashSet();
+    try {
+      while(true) {
+        in.addAll(putEvents(channel, "will-not-restart", 1, 1));
+      }
+    } catch (ChannelException e) {
+      Assert.assertEquals("Cannot acquire capacity. [channel="
+          +channel.getName()+"]", e.getMessage());
+    }
+    channel.stop();
+    Map<String, String> noEncryptionOverrides = getOverrides();
+    channel = TestUtils.createFileChannel(checkpointDir.getAbsolutePath(),
+        dataDir, noEncryptionOverrides);
+    channel.start();
+
+    if(channel.isOpen()) {
+      try {
+        takeEvents(channel, 1, 1);
+        Assert.fail("Channel was opened and take did not throw exception");
+      } catch(ChannelException ex) {
+        // expected
+      }
+    }
+  }
+  @Test
+  public void testUnencyrptedAndEncryptedLogs() throws Exception {
+    Map<String, String> noEncryptionOverrides = getOverrides();
+    channel = TestUtils.createFileChannel(checkpointDir.getAbsolutePath(),
+        dataDir, noEncryptionOverrides);
+    channel.start();
+    Assert.assertTrue(channel.isOpen());
+    Set<String> in = Sets.newHashSet();
+    try {
+      while(true) {
+        in.addAll(putEvents(channel, "unencrypted-and-encrypted", 1, 1));
+      }
+    } catch (ChannelException e) {
+      Assert.assertEquals("Cannot acquire capacity. [channel="
+          +channel.getName()+"]", e.getMessage());
+    }
+    int numEventsToRemove = in.size() / 2;
+    for (int i = 0; i < numEventsToRemove; i++) {
+      Assert.assertTrue(in.removeAll(takeEvents(channel, 1, 1)));
+    }
+    // now we have logs with no encryption and the channel is half full
+    channel.stop();
+    Map<String, String> overrides = getOverridesForEncryption();
+    channel = TestUtils.createFileChannel(checkpointDir.getAbsolutePath(),
+        dataDir, overrides);
+    channel.start();
+    Assert.assertTrue(channel.isOpen());
+    try {
+      while(true) {
+        in.addAll(putEvents(channel, "unencrypted-and-encrypted", 1, 1));
+      }
+    } catch (ChannelException e) {
+      Assert.assertEquals("Cannot acquire capacity. [channel="
+          +channel.getName()+"]", e.getMessage());
+    }
+    Set<String> out = takeEvents(channel, 1, Integer.MAX_VALUE);
+    compareInputAndOut(in, out);
+  }
+  @Test
+  public void testBadKeyProviderInvalidValue() throws Exception {
+    Map<String, String> overrides = getOverridesForEncryption();
+    overrides.put(EncryptionConfiguration.ENCRYPTION_PREFIX + "." +
+        EncryptionConfiguration.KEY_PROVIDER, "invalid");
+    try {
+      channel = TestUtils.createFileChannel(checkpointDir.getAbsolutePath(),
+          dataDir, overrides);
+      Assert.fail();
+    } catch(FlumeException ex) {
+      Assert.assertEquals("java.lang.ClassNotFoundException: invalid",
+          ex.getMessage());
+    }
+  }
+  @Test
+  public void testBadKeyProviderInvalidClass() throws Exception {
+    Map<String, String> overrides = getOverridesForEncryption();
+    overrides.put(EncryptionConfiguration.ENCRYPTION_PREFIX + "." +
+        EncryptionConfiguration.KEY_PROVIDER, String.class.getName());
+    try {
+      channel = TestUtils.createFileChannel(checkpointDir.getAbsolutePath(),
+          dataDir, overrides);
+      Assert.fail();
+    } catch(FlumeException ex) {
+      Assert.assertEquals("Unable to instantiate Builder from java.lang.String",
+          ex.getMessage());
+    }
+  }
+  @Test
+  public void testBadCipherProviderInvalidValue() throws Exception {
+    Map<String, String> overrides = getOverridesForEncryption();
+    overrides.put(EncryptionConfiguration.ENCRYPTION_PREFIX + "." +
+        EncryptionConfiguration.CIPHER_PROVIDER, "invalid");
+    channel = TestUtils.createFileChannel(checkpointDir.getAbsolutePath(),
+        dataDir, overrides);
+    channel.start();
+    Assert.assertFalse(channel.isOpen());
+  }
+  @Test
+  public void testBadCipherProviderInvalidClass() throws Exception {
+    Map<String, String> overrides = getOverridesForEncryption();
+    overrides.put(EncryptionConfiguration.ENCRYPTION_PREFIX + "." +
+        EncryptionConfiguration.CIPHER_PROVIDER, String.class.getName());
+    channel = TestUtils.createFileChannel(checkpointDir.getAbsolutePath(),
+        dataDir, overrides);
+    channel.start();
+    Assert.assertFalse(channel.isOpen());
+  }
+  @Test
+  public void testMissingKeyStoreFile() throws Exception {
+    Map<String, String> overrides = getOverridesForEncryption();
+    overrides.put(EncryptionConfiguration.ENCRYPTION_PREFIX + "." +
+        EncryptionConfiguration.JCE_FILE_KEY_STORE_FILE, "/path/does/not/exist");
+    try {
+      channel = TestUtils.createFileChannel(checkpointDir.getAbsolutePath(),
+          dataDir, overrides);
+      Assert.fail();
+    } catch(RuntimeException ex) {
+      Assert.assertEquals("java.io.FileNotFoundException: /path/does/not/exist " +
+          "(No such file or directory)", ex.getMessage());
+    }
+  }
+  @Test
+  public void testMissingKeyStorePasswordFile() throws Exception {
+    Map<String, String> overrides = getOverridesForEncryption();
+    overrides.put(EncryptionConfiguration.ENCRYPTION_PREFIX + "." +
+        EncryptionConfiguration.JCE_FILE_KEY_STORE_PASSWORD_FILE, "/path/does/not/exist");
+    try {
+      channel = TestUtils.createFileChannel(checkpointDir.getAbsolutePath(),
+          dataDir, overrides);
+      Assert.fail();
+    } catch(RuntimeException ex) {
+      Assert.assertEquals("java.io.FileNotFoundException: /path/does/not/exist " +
+          "(No such file or directory)", ex.getMessage());
+    }
+  }
+  @Test
+  public void testBadKeyStorePassword() throws Exception {
+    Files.write("invalid", keyStorePasswordFile, Charsets.UTF_8);
+    Map<String, String> overrides = getOverridesForEncryption();
+    try {
+      channel = TestUtils.createFileChannel(checkpointDir.getAbsolutePath(),
+          dataDir, overrides);
+      Assert.fail();
+    } catch(RuntimeException ex) {
+      Assert.assertEquals("java.io.IOException: Keystore was tampered with, or " +
+          "password was incorrect", ex.getMessage());
+    }
+  }
+  @Test
+  public void testBadKeyAlias() throws Exception {
+    Map<String, String> overrides = getOverridesForEncryption();
+    overrides.put(EncryptionConfiguration.ENCRYPTION_PREFIX + "." +
+        EncryptionConfiguration.KEY_ALIAS, "invalid");
+    channel = TestUtils.createFileChannel(checkpointDir.getAbsolutePath(),
+        dataDir, overrides);
+    channel.start();
+    Assert.assertFalse(channel.isOpen());
+  }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/encryption/TestJCEFileKeyProvider.java
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/encryption/TestJCEFileKeyProvider.java b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/encryption/TestJCEFileKeyProvider.java
new file mode 100644
index 0000000..519952e
--- /dev/null
+++ b/flume-ng-channels/flume-file-channel/src/test/java/org/apache/flume/channel/file/encryption/TestJCEFileKeyProvider.java
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.flume.channel.file.encryption;
+
+import java.io.File;
+import java.security.Key;
+import java.util.Map;
+
+import junit.framework.Assert;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.flume.Context;
+import org.apache.flume.channel.file.TestUtils;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import com.google.common.base.Charsets;
+import com.google.common.collect.Maps;
+import com.google.common.io.Files;
+
+public class TestJCEFileKeyProvider {
+  private CipherProvider.Encryptor encryptor;
+  private CipherProvider.Decryptor decryptor;
+  private File baseDir;
+  private File keyStoreFile;
+  private File keyStorePasswordFile;
+  private Map<String, File> keyAliasPassword;
+
+  @Before
+  public void setup() throws Exception {
+    baseDir = Files.createTempDir();
+    keyStorePasswordFile = new File(baseDir, "keyStorePasswordFile");
+    Files.write("keyStorePassword", keyStorePasswordFile, Charsets.UTF_8);
+    keyAliasPassword = Maps.newHashMap();
+    keyStoreFile = new File(baseDir, "keyStoreFile");
+    Assert.assertTrue(keyStoreFile.createNewFile());
+
+  }
+  @After
+  public void cleanup() {
+    FileUtils.deleteQuietly(baseDir);
+  }
+  private void initializeForKey(Key key) {
+    encryptor =
+        new AESCTRNoPaddingProvider.EncryptorBuilder().setKey(key).build();
+    decryptor =
+        new AESCTRNoPaddingProvider.DecryptorBuilder()
+        .setKey(key).setParameters(encryptor.getParameters()).build();
+  }
+  @Test
+  public void testWithNewKeyStore() throws Exception {
+    createNewKeyStore();
+    EncryptionTestUtils.createKeyStore(keyStoreFile, keyStorePasswordFile,
+        keyAliasPassword);
+    Context context = new Context(EncryptionTestUtils.
+        configureForKeyStore(keyStoreFile, keyStorePasswordFile,
+            keyAliasPassword));
+    KeyProvider keyProvider = KeyProviderFactory.
+        getInstance(KeyProviderType.JCEKSFILE.name(), context);
+    testKeyProvider(keyProvider);
+  }
+  @Test
+  public void testWithExistingKeyStore() throws Exception {
+    keyAliasPassword.putAll(EncryptionTestUtils.
+        configureTestKeyStore(baseDir, keyStoreFile));
+    Context context = new Context(EncryptionTestUtils.
+        configureForKeyStore(keyStoreFile, keyStorePasswordFile,
+            keyAliasPassword));
+    KeyProvider keyProvider = KeyProviderFactory.
+        getInstance(KeyProviderType.JCEKSFILE.name(), context);
+    testKeyProvider(keyProvider);
+  }
+  private void createNewKeyStore() throws Exception {
+    for(int i = 0; i < 10; i++) {
+      // create some with passwords, some without
+      if(i % 2 == 0) {
+        String alias = "test-" + i;
+        String password = String.valueOf(i);
+        keyAliasPassword.put(alias,
+            TestUtils.writeStringToFile(baseDir, alias, password));
+      }
+    }
+  }
+  private void testKeyProvider(KeyProvider keyProvider) {
+    for(String alias : keyAliasPassword.keySet()) {
+      Key key = keyProvider.getKey(alias);
+      initializeForKey(key);
+      String expected = "some text here " + alias;
+      byte[] cipherText = encryptor.
+          encrypt(expected.getBytes(Charsets.UTF_8));
+      byte[] clearText = decryptor.decrypt(cipherText);
+      Assert.assertEquals(expected, new String(clearText, Charsets.UTF_8));
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/flume/blob/d3c85b1d/flume-ng-channels/flume-file-channel/src/test/resources/test.keystore
----------------------------------------------------------------------
diff --git a/flume-ng-channels/flume-file-channel/src/test/resources/test.keystore b/flume-ng-channels/flume-file-channel/src/test/resources/test.keystore
new file mode 100644
index 0000000..ddaa95d
Binary files /dev/null and b/flume-ng-channels/flume-file-channel/src/test/resources/test.keystore differ