make encryption and decryption tolerant to missing key

Today, an exception is thrown when calls to the encrypt and decrypt methods are
made without a key being present. For now, we will not require the system key and
this behavior is undesirable.

This commit changes the behavior to just return the provided characters or bytes
when no key is present. Additionally, a method has been added for callers to see
if encryption is supported. Finally, the listener interface has been made public and
expanded to provide the old keys when the keys are changed. This allows
consumers to decrypt with the old key and re-encrypt with the new key.

Original commit: elastic/x-pack-elasticsearch@de3d5b6180

src/main/java/org/elasticsearch/shield/authc/support/CharArrays.java

@@ -44,4 +44,22 @@ public class CharArrays {
         Arrays.fill(byteBuffer.array(), (byte) 0); // clear sensitive data
         return bytes;
     }
+
+    public static boolean charsBeginsWith(String prefix, char[] chars) {
+        if (chars == null || prefix == null) {
+            return false;
+        }
+
+        if (prefix.length() > chars.length) {
+            return false;
+        }
+
+        for (int i = 0; i < prefix.length(); i++) {
+            if (chars[i] != prefix.charAt(i)) {
+                return false;
+            }
+        }
+
+        return true;
+    }
 }

src/main/java/org/elasticsearch/shield/crypto/CryptoService.java

@@ -5,6 +5,8 @@
  */
 package org.elasticsearch.shield.crypto;
 
+import javax.crypto.SecretKey;
+
 /**
  * Service that provides cryptographic methods based on a shared system key
  */
@@ -12,25 +14,41 @@ public interface CryptoService {
 
     /**
      * Signs the given text and returns the signed text (original text + signature)
+     * @param text the string to sign
      */
     String sign(String text);
 
     /**
      * Unsigns the given signed text, verifies the original text with the attached signature and if valid returns
      * the unsigned (original) text. If signature verification fails a {@link SignatureException} is thrown.
+     * @param text the string to unsign and verify
      */
     String unsignAndVerify(String text);
 
+    /**
+     * Signs the given text and returns the signed text (original text + signature)
+     * @param text the string to sign
+     * @param key the key to sign the text with
+     */
+    String sign(String text, SecretKey key);
+
+    /**
+     * Unsigns the given signed text, verifies the original text with the attached signature and if valid returns
+     * the unsigned (original) text. If signature verification fails a {@link SignatureException} is thrown.
+     * @param text the string to unsign and verify
+     * @param key the key to unsign the text with
+     */
+    String unsignAndVerify(String text, SecretKey key);
+
     /**
      * Checks whether the given text is signed.
      */
     boolean signed(String text);
 
     /**
-     * Encrypts the provided char array and returns the encrypted values in a Base64 encoded char array
+     * Encrypts the provided char array and returns the encrypted values in a char array
      * @param chars the characters to encrypt
-     * @return Base64 character array representing the encrypted data
-     * @throws UnsupportedOperationException if the system key is not present
+     * @return character array representing the encrypted data
      */
     char[] encrypt(char[] chars);
 
@@ -38,23 +56,74 @@ public interface CryptoService {
      * Encrypts the provided byte array and returns the encrypted value
      * @param bytes the data to encrypt
      * @return encrypted data
-     * @throws UnsupportedOperationException if the system key is not present
      */
     byte[] encrypt(byte[] bytes);
 
     /**
      * Decrypts the provided char array and returns the plain-text chars
-     * @param chars the Base64 encoded data to decrypt
+     * @param chars the data to decrypt
      * @return plaintext chars
-     * @throws UnsupportedOperationException if the system key is not present
      */
     char[] decrypt(char[] chars);
 
+    /**
+     * Decrypts the provided char array and returns the plain-text chars
+     * @param chars the data to decrypt
+     * @param key the key to decrypt the data with
+     * @return plaintext chars
+     */
+    char[] decrypt(char[] chars, SecretKey key);
+
     /**
      * Decrypts the provided byte array and returns the unencrypted bytes
      * @param bytes the bytes to decrypt
      * @return plaintext bytes
-     * @throws UnsupportedOperationException if the system key is not present
      */
     byte[] decrypt(byte[] bytes);
+
+    /**
+     * Decrypts the provided byte array and returns the unencrypted bytes
+     * @param bytes the bytes to decrypt
+     * @param key the key to decrypt the data with
+     * @return plaintext bytes
+     */
+    byte[] decrypt(byte[] bytes, SecretKey key);
+
+    /**
+     * Checks whether the given chars are encrypted
+     * @param chars the chars to check if they are encrypted
+     * @return true is data is encrypted
+     */
+    boolean encrypted(char[] chars);
+
+    /**
+     * Checks whether the given bytes are encrypted
+     * @param bytes the chars to check if they are encrypted
+     * @return true is data is encrypted
+     */
+    boolean encrypted(byte[] bytes);
+
+    /**
+     * Registers a listener to be notified of key changes
+     * @param listener the listener to be notified
+     */
+    void register(Listener listener);
+
+    /**
+     * Flag for callers to determine if values will actually be encrypted or returned plaintext
+     * @return true if values will be encrypted
+     */
+    boolean encryptionEnabled();
+
+    interface Listener {
+        /**
+         * This method will be called immediately after a new system key and encryption key are loaded by the
+         * service. This provides the old keys back to the clients so that they may perform decryption and re-encryption
+         * of data after a key has been changed
+         *
+         * @param oldSystemKey
+         * @param oldEncryptionKey
+         */
+        void onKeyChange(SecretKey oldSystemKey, SecretKey oldEncryptionKey);
+    }
 }

src/main/java/org/elasticsearch/shield/crypto/InternalCryptoService.java

@@ -30,6 +30,9 @@ import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
 import java.security.SecureRandom;
 import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.regex.Pattern;
 
 import static org.elasticsearch.shield.authc.support.SecuredString.constantTimeEquals;
@@ -47,13 +50,15 @@ public class InternalCryptoService extends AbstractLifecycleComponent<InternalCr
     static final String HMAC_ALGO = "HmacSHA1";
     static final String DEFAULT_ENCRYPTION_ALGORITHM = "AES/CTR/NoPadding";
     static final String DEFAULT_KEY_ALGORITH = "AES";
+    static final String ENCRYPTED_TEXT_PREFIX = "::es_encrypted::";
+    static final byte[] ENCRYPTED_BYTE_PREFIX = ENCRYPTED_TEXT_PREFIX.getBytes(Charsets.UTF_8);
     static final int DEFAULT_KEY_LENGTH = 128;
 
     private static final Pattern SIG_PATTERN = Pattern.compile("^\\$\\$[0-9]+\\$\\$.+");
 
     private final Environment env;
     private final ResourceWatcherService watcherService;
-    private final Listener listener;
+    private final List<Listener> listeners;
     private final SecureRandom secureRandom = new SecureRandom();
     private final String encryptionAlgorithm;
     private final String keyAlgorithm;
@@ -67,14 +72,14 @@ public class InternalCryptoService extends AbstractLifecycleComponent<InternalCr
 
     @Inject
     public InternalCryptoService(Settings settings, Environment env, ResourceWatcherService watcherService) {
-        this(settings, env, watcherService, Listener.NOOP);
+        this(settings, env, watcherService, Collections.<Listener>emptyList());
     }
 
-    InternalCryptoService(Settings settings, Environment env, ResourceWatcherService watcherService, Listener listener) {
+    InternalCryptoService(Settings settings, Environment env, ResourceWatcherService watcherService, List<Listener> listeners) {
         super(settings);
         this.env = env;
         this.watcherService = watcherService;
-        this.listener = listener;
+        this.listeners = new CopyOnWriteArrayList<>(listeners);
         this.encryptionAlgorithm = settings.get("shield.encryption.algorithm", DEFAULT_ENCRYPTION_ALGORITHM);
         this.keyLength = settings.getAsInt("shield.encryption_key.length", DEFAULT_KEY_LENGTH);
         if (keyLength % 8 != 0) {
@@ -90,7 +95,7 @@ public class InternalCryptoService extends AbstractLifecycleComponent<InternalCr
         systemKey = readSystemKey(keyFile);
         encryptionKey = encryptionKey(systemKey, keyLength, keyAlgorithm);
         FileWatcher watcher = new FileWatcher(keyFile.getParent());
-        watcher.addListener(new FileListener(listener));
+        watcher.addListener(new FileListener(listeners));
         try {
             watcherService.add(watcher, ResourceWatcherService.Frequency.HIGH);
         } catch (IOException e) {
@@ -132,17 +137,25 @@ public class InternalCryptoService extends AbstractLifecycleComponent<InternalCr
 
     @Override
     public String sign(String text) {
-        SecretKey key = this.systemKey;
+        return sign(text, this.systemKey);
+    }
+
+    @Override
+    public String sign(String text, SecretKey key) {
         if (key == null) {
             return text;
         }
-        String sigStr = signInternal(text);
+        String sigStr = signInternal(text, key);
         return "$$" + sigStr.length() + "$$" + sigStr + text;
     }
 
     @Override
     public String unsignAndVerify(String signedText) {
-        SecretKey key = this.systemKey;
+        return unsignAndVerify(signedText, this.systemKey);
+    }
+
+    @Override
+    public String unsignAndVerify(String signedText, SecretKey key) {
         if (key == null) {
             return signedText;
         }
@@ -165,7 +178,7 @@ public class InternalCryptoService extends AbstractLifecycleComponent<InternalCr
         }
 
         try {
-            String sig = signInternal(text);
+            String sig = signInternal(text, key);
             if (constantTimeEquals(sig, receivedSignature)) {
                 return text;
             }
@@ -186,50 +199,94 @@ public class InternalCryptoService extends AbstractLifecycleComponent<InternalCr
     public char[] encrypt(char[] chars) {
         SecretKey key = this.encryptionKey;
         if (key == null) {
-            throw new UnsupportedOperationException("encryption cannot be performed without a system key. please run bin/shield/syskeygen on one node and copy\n"
-                + "the file [" + ShieldPlugin.resolveConfigFile(env, FILE_NAME) + "] to all nodes and the key will be loaded automatically.");
+            logger.warn("encrypt called without a key, returning plain text. run syskeygen and copy same key to all nodes to enable encryption");
+            return chars;
         }
 
         byte[] charBytes = CharArrays.toUtf8Bytes(chars);
-        return Base64.encodeBytes(encryptInternal(charBytes, key)).toCharArray();
+        String base64 = Base64.encodeBytes(encryptInternal(charBytes, key));
+        return ENCRYPTED_TEXT_PREFIX.concat(base64).toCharArray();
     }
 
     @Override
     public byte[] encrypt(byte[] bytes) {
         SecretKey key = this.encryptionKey;
         if (key == null) {
-            throw new UnsupportedOperationException("encryption cannot be performed without a system key. please run bin/shield/syskeygen on one node and copy\n"
-                    + "the file [" + ShieldPlugin.resolveConfigFile(env, FILE_NAME) + "] to all nodes and the key will be loaded automatically.");
+            logger.warn("encrypt called without a key, returning plain text. run syskeygen and copy same key to all nodes to enable encryption");
+            return bytes;
         }
-        return encryptInternal(bytes, key);
+        byte[] encrypted = encryptInternal(bytes, key);
+        byte[] prefixed = new byte[ENCRYPTED_BYTE_PREFIX.length + encrypted.length];
+        System.arraycopy(ENCRYPTED_BYTE_PREFIX, 0, prefixed, 0, ENCRYPTED_BYTE_PREFIX.length);
+        System.arraycopy(encrypted, 0, prefixed, ENCRYPTED_BYTE_PREFIX.length, encrypted.length);
+        return prefixed;
     }
 
     @Override
     public char[] decrypt(char[] chars) {
-        SecretKey key = this.encryptionKey;
+        return decrypt(chars, this.encryptionKey);
+    }
+
+    @Override
+    public char[] decrypt(char[] chars, SecretKey key) {
         if (key == null) {
-            throw new UnsupportedOperationException("decryption cannot be performed without a system key. please run bin/shield/syskeygen on one node and copy\n"
-                    + "the file [" + ShieldPlugin.resolveConfigFile(env, FILE_NAME) + "] to all nodes and the key will be loaded automatically.");
+            return chars;
         }
 
+        if (!encrypted(chars)) {
+            // Not encrypted
+            return chars;
+        }
+
+        String encrypted = new String(chars, ENCRYPTED_TEXT_PREFIX.length(), chars.length - ENCRYPTED_TEXT_PREFIX.length());
         byte[] bytes;
         try {
-            bytes = Base64.decode(new String(chars));
+            bytes = Base64.decode(encrypted);
         } catch (IOException e) {
             throw new ShieldException("unable to decode encrypted data", e);
         }
+
         byte[] decrypted = decryptInternal(bytes, key);
         return CharArrays.utf8BytesToChars(decrypted);
     }
 
     @Override
     public byte[] decrypt(byte[] bytes) {
-        SecretKey key = this.encryptionKey;
+        return decrypt(bytes, this.encryptionKey);
+    }
+
+    @Override
+    public byte[] decrypt(byte[] bytes, SecretKey key) {
         if (key == null) {
-            throw new UnsupportedOperationException("decryption cannot be performed without a system key. please run bin/shield/syskeygen on one node and copy\n"
-                    + "the file [" + ShieldPlugin.resolveConfigFile(env, FILE_NAME) + "] to all nodes and the key will be loaded automatically.");
+            return bytes;
         }
-        return decryptInternal(bytes, key);
+
+        if (!encrypted(bytes)) {
+            return bytes;
+        }
+
+        byte[] encrypted = Arrays.copyOfRange(bytes, ENCRYPTED_BYTE_PREFIX.length, bytes.length);
+        return decryptInternal(encrypted, key);
+    }
+
+    @Override
+    public boolean encrypted(char[] chars) {
+        return CharArrays.charsBeginsWith(ENCRYPTED_TEXT_PREFIX, chars);
+    }
+
+    @Override
+    public boolean encrypted(byte[] bytes) {
+        return bytesBeginsWith(ENCRYPTED_BYTE_PREFIX, bytes);
+    }
+
+    @Override
+    public void register(Listener listener) {
+        this.listeners.add(listener);
+    }
+
+    @Override
+    public boolean encryptionEnabled() {
+        return this.encryptionKey != null;
     }
 
     private byte[] encryptInternal(byte[] bytes, SecretKey key) {
@@ -276,8 +333,8 @@ public class InternalCryptoService extends AbstractLifecycleComponent<InternalCr
         }
     }
 
-    private String signInternal(String text) {
-        Mac mac = createMac(systemKey);
+    private static String signInternal(String text, SecretKey key) {
+        Mac mac = createMac(key);
         byte[] sig = mac.doFinal(text.getBytes(Charsets.UTF_8));
         try {
             return Base64.encodeBytes(sig, 0, sig.length, Base64.URL_SAFE);
@@ -323,54 +380,99 @@ public class InternalCryptoService extends AbstractLifecycleComponent<InternalCr
         }
     }
 
+    private static boolean bytesBeginsWith(byte[] prefix, byte[] bytes) {
+        if (bytes == null || prefix == null) {
+            return false;
+        }
+
+        if (prefix.length > bytes.length) {
+            return false;
+        }
+
+        for (int i = 0; i < prefix.length; i++) {
+            if (bytes[i] != prefix[i]) {
+                return false;
+            }
+        }
+
+        return true;
+    }
+
     private class FileListener extends FileChangesListener {
 
-        private final Listener listener;
+        private final List<Listener> listeners;
 
-        private FileListener(Listener listener) {
-            this.listener = listener;
+        private FileListener(List<Listener> listeners) {
+            this.listeners = listeners;
         }
 
         @Override
         public void onFileCreated(Path file) {
             if (file.equals(keyFile)) {
+                final SecretKey oldSystemKey = systemKey;
+                final SecretKey oldEncryptionKey = encryptionKey;
+
                 systemKey = readSystemKey(file);
                 encryptionKey = encryptionKey(systemKey, keyLength, keyAlgorithm);
                 logger.info("system key [{}] has been loaded", file.toAbsolutePath());
-                listener.onKeyRefresh();
+
+                callListeners(oldSystemKey, oldEncryptionKey);
             }
         }
 
         @Override
         public void onFileDeleted(Path file) {
             if (file.equals(keyFile)) {
+                final SecretKey oldSystemKey = systemKey;
+                final SecretKey oldEncryptionKey = encryptionKey;
                 logger.error("system key file was removed! as long as the system key file is missing, elasticsearch " +
-                        "won't function as expected for some requests (e.g. scroll/scan) and won't be able to decrypt\n" +
-                        "previously encrypted values without the original key");
+                        "won't function as expected for some requests (e.g. scroll/scan)");
                 systemKey = null;
                 encryptionKey = null;
+
+                callListeners(oldSystemKey, oldEncryptionKey);
             }
         }
 
         @Override
         public void onFileChanged(Path file) {
             if (file.equals(keyFile)) {
-                logger.warn("system key file changed! previously encrypted values cannot be successfully decrypted with a different key");
+                final SecretKey oldSystemKey = systemKey;
+                final SecretKey oldEncryptionKey = encryptionKey;
+
+                logger.warn("system key file changed!");
                 systemKey = readSystemKey(file);
                 encryptionKey = encryptionKey(systemKey, keyLength, keyAlgorithm);
-                listener.onKeyRefresh();
+
+                callListeners(oldSystemKey, oldEncryptionKey);
+            }
+        }
+
+        private void callListeners(SecretKey oldSystemKey, SecretKey oldEncryptionKey) {
+            Throwable th = null;
+            for (Listener listener : listeners) {
+                try {
+                    listener.onKeyChange(oldSystemKey, oldEncryptionKey);
+                } catch (Throwable t) {
+                    if (th == null) {
+                        th = t;
+                    } else {
+                        th.addSuppressed(t);
+                    }
+                }
+            }
+
+            // all listeners were notified now rethrow
+            if (th != null) {
+                logger.error("called all key change listeners but one or more exceptions was thrown", th);
+                if (th instanceof RuntimeException) {
+                    throw (RuntimeException) th;
+                } else if (th instanceof Error) {
+                    throw (Error) th;
+                } else {
+                    throw new RuntimeException(th);
+                }
             }
         }
     }
-
-    static interface Listener {
-
-        final Listener NOOP = new Listener() {
-            @Override
-            public void onKeyRefresh() {
-            }
-        };
-
-        void onKeyRefresh();
-    }
 }

src/test/java/org/elasticsearch/shield/crypto/InternalCryptoServiceTests.java

@@ -5,6 +5,7 @@
  */
 package org.elasticsearch.shield.crypto;
 
+import org.elasticsearch.common.base.Charsets;
 import org.elasticsearch.common.io.Streams;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.env.Environment;
@@ -18,7 +19,9 @@ import org.junit.Test;
 import java.io.OutputStream;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import javax.crypto.SecretKey;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
@@ -154,6 +157,7 @@ public class InternalCryptoServiceTests extends ElasticsearchTestCase {
     @Test
     public void testEncryptionAndDecryptionChars() {
         InternalCryptoService service = new InternalCryptoService(settings, env, watcherService).start();
+        assertThat(service.encryptionEnabled(), is(true));
         final char[] chars = randomAsciiOfLengthBetween(0, 1000).toCharArray();
         final char[] encrypted = service.encrypt(chars);
         assertThat(encrypted, notNullValue());
@@ -166,6 +170,7 @@ public class InternalCryptoServiceTests extends ElasticsearchTestCase {
     @Test
     public void testEncryptionAndDecryptionBytes() {
         InternalCryptoService service = new InternalCryptoService(settings, env, watcherService).start();
+        assertThat(service.encryptionEnabled(), is(true));
         final byte[] bytes = randomByteArray();
         final byte[] encrypted = service.encrypt(bytes);
         assertThat(encrypted, notNullValue());
@@ -178,53 +183,48 @@ public class InternalCryptoServiceTests extends ElasticsearchTestCase {
     @Test
     public void testEncryptionAndDecryptionCharsWithoutKey() {
         InternalCryptoService service = new InternalCryptoService(Settings.EMPTY, env, watcherService).start();
+        assertThat(service.encryptionEnabled(), is(false));
         final char[] chars = randomAsciiOfLengthBetween(0, 1000).toCharArray();
-        try {
-            service.encrypt(chars);
-            fail("exception should have been thrown");
-        } catch (Exception e) {
-            assertThat(e, instanceOf(UnsupportedOperationException.class));
-            assertThat(e.getMessage(), containsString("system_key"));
-        }
-
-        try {
-            service.decrypt(chars);
-        } catch (Exception e) {
-            assertThat(e, instanceOf(UnsupportedOperationException.class));
-            assertThat(e.getMessage(), containsString("system_key"));
-        }
+        final char[] encryptedChars = service.encrypt(chars);
+        final char[] decryptedChars = service.decrypt(encryptedChars);
+        assertThat(chars, equalTo(encryptedChars));
+        assertThat(chars, equalTo(decryptedChars));
     }
 
     @Test
     public void testEncryptionAndDecryptionBytesWithoutKey() {
         InternalCryptoService service = new InternalCryptoService(Settings.EMPTY, env, watcherService).start();
+        assertThat(service.encryptionEnabled(), is(false));
         final byte[] bytes = randomByteArray();
-        try {
-            service.encrypt(bytes);
-            fail("exception should have been thrown");
-        } catch (Exception e) {
-            assertThat(e, instanceOf(UnsupportedOperationException.class));
-            assertThat(e.getMessage(), containsString("system_key"));
-        }
+        final byte[] encryptedBytes = service.encrypt(bytes);
+        final byte[] decryptedBytes = service.decrypt(bytes);
+        assertThat(bytes, equalTo(encryptedBytes));
+        assertThat(decryptedBytes, equalTo(encryptedBytes));
+    }
 
-        try {
-            service.decrypt(bytes);
-        } catch (Exception e) {
-            assertThat(e, instanceOf(UnsupportedOperationException.class));
-            assertThat(e.getMessage(), containsString("system_key"));
-        }
+    @Test
+    public void testEncryptionEnabledWithKey() {
+        InternalCryptoService service = new InternalCryptoService(settings, env, watcherService).start();
+        assertThat(service.encryptionEnabled(), is(true));
+    }
+
+    @Test
+    public void testEncryptionEnabledWithoutKey() {
+        InternalCryptoService service = new InternalCryptoService(Settings.EMPTY, env, watcherService).start();
+        assertThat(service.encryptionEnabled(), is(false));
     }
 
     @Test
     public void testChangingAByte() {
         InternalCryptoService service = new InternalCryptoService(settings, env, watcherService).start();
+        assertThat(service.encryptionEnabled(), is(true));
         // We need at least one byte to test changing a byte, otherwise output is always the same
         final byte[] bytes = randomByteArray(1);
         final byte[] encrypted = service.encrypt(bytes);
         assertThat(encrypted, notNullValue());
         assertThat(Arrays.equals(encrypted, bytes), is(false));
 
-        int tamperedIndex = randomIntBetween(0, encrypted.length - 1);
+        int tamperedIndex = randomIntBetween(InternalCryptoService.ENCRYPTED_BYTE_PREFIX.length, encrypted.length - 1);
         final byte untamperedByte = encrypted[tamperedIndex];
         byte tamperedByte = randomByte();
         while (tamperedByte == untamperedByte) {
@@ -235,22 +235,57 @@ public class InternalCryptoServiceTests extends ElasticsearchTestCase {
         assertThat(Arrays.equals(bytes, decrypted), is(false));
     }
 
+    @Test
+    public void testEncryptedChar() {
+        InternalCryptoService service = new InternalCryptoService(settings, env, watcherService).start();
+        assertThat(service.encryptionEnabled(), is(true));
+
+        assertThat(service.encrypted((char[]) null), is(false));
+        assertThat(service.encrypted(new char[0]), is(false));
+        assertThat(service.encrypted(new char[InternalCryptoService.ENCRYPTED_TEXT_PREFIX.length()]), is(false));
+        assertThat(service.encrypted(InternalCryptoService.ENCRYPTED_TEXT_PREFIX.toCharArray()), is(true));
+        assertThat(service.encrypted(randomAsciiOfLengthBetween(0, 100).toCharArray()), is(false));
+        assertThat(service.encrypted(service.encrypt(randomAsciiOfLength(10).toCharArray())), is(true));
+    }
+
+    @Test
+    public void testEncryptedByte() {
+        InternalCryptoService service = new InternalCryptoService(settings, env, watcherService).start();
+        assertThat(service.encryptionEnabled(), is(true));
+
+        assertThat(service.encrypted((byte[]) null), is(false));
+        assertThat(service.encrypted(new byte[0]), is(false));
+        assertThat(service.encrypted(new byte[InternalCryptoService.ENCRYPTED_BYTE_PREFIX.length]), is(false));
+        assertThat(service.encrypted(InternalCryptoService.ENCRYPTED_BYTE_PREFIX), is(true));
+        assertThat(service.encrypted(randomAsciiOfLengthBetween(0, 100).getBytes(Charsets.UTF_8)), is(false));
+        assertThat(service.encrypted(service.encrypt(randomAsciiOfLength(10).getBytes(Charsets.UTF_8))), is(true));
+    }
+
     @Test
     public void testReloadKey() throws Exception {
         final CountDownLatch latch = new CountDownLatch(1);
-        InternalCryptoService service = new InternalCryptoService(settings, env, watcherService, new InternalCryptoService.Listener() {
+        final CryptoService.Listener listener = new CryptoService.Listener() {
             @Override
-            public void onKeyRefresh() {
+            public void onKeyChange(SecretKey oldSystemKey, SecretKey oldEncryptionKey) {
                 latch.countDown();
             }
-        }).start();
+        };
+
+        // randomize how we set the listener
+        InternalCryptoService service;
+        if (randomBoolean()) {
+            service = new InternalCryptoService(settings, env, watcherService, Collections.singletonList(listener)).start();
+        } else {
+            service = new InternalCryptoService(settings, env, watcherService).start();
+            service.register(listener);
+        }
 
         String text = randomAsciiOfLength(10);
         String signed = service.sign(text);
         char[] textChars = text.toCharArray();
         char[] encrypted = service.encrypt(textChars);
 
-        // we need to sleep so to ensure the timestamp of the file will definitely change
+        // we need to sleep to ensure the timestamp of the file will definitely change
         // and so the resource watcher will pick up the change.
         Thread.sleep(1000L);
 
@@ -271,6 +306,135 @@ public class InternalCryptoServiceTests extends ElasticsearchTestCase {
         assertThat(Arrays.equals(textChars, decrypted2), is(true));
     }
 
+    @Test
+    public void testReencryptValuesOnKeyChange() throws Exception {
+        final InternalCryptoService service = new InternalCryptoService(settings, env, watcherService).start();
+        assertThat(service.encryptionEnabled(), is(true));
+        final char[] text = randomAsciiOfLength(10).toCharArray();
+        final char[] encrypted = service.encrypt(text);
+        assertThat(text, not(equalTo(encrypted)));
+
+        final CountDownLatch latch = new CountDownLatch(1);
+        service.register(new CryptoService.Listener() {
+            @Override
+            public void onKeyChange(SecretKey oldSystemKey, SecretKey oldEncryptionKey) {
+                final char[] plainText = service.decrypt(encrypted, oldEncryptionKey);
+                assertThat(plainText, equalTo(text));
+                final char[] newEncrypted = service.encrypt(plainText);
+                assertThat(newEncrypted, not(equalTo(encrypted)));
+                assertThat(newEncrypted, not(equalTo(plainText)));
+                latch.countDown();
+            }
+        });
+
+        // we need to sleep to ensure the timestamp of the file will definitely change
+        // and so the resource watcher will pick up the change.
+        Thread.sleep(1000);
+
+        Files.write(keyFile, InternalCryptoService.generateKey());
+        if (!latch.await(10, TimeUnit.SECONDS)) {
+            fail("waiting too long for test to complete. Expected callback is not called or finished running");
+        }
+    }
+
+    @Test
+    public void testResignValuesOnKeyChange() throws Exception {
+        final InternalCryptoService service = new InternalCryptoService(settings, env, watcherService).start();
+        final String text = randomAsciiOfLength(10);
+        final String signed = service.sign(text);
+        assertThat(text, not(equalTo(signed)));
+
+        final CountDownLatch latch = new CountDownLatch(1);
+        service.register(new CryptoService.Listener() {
+            @Override
+            public void onKeyChange(SecretKey oldSystemKey, SecretKey oldEncryptionKey) {
+                assertThat(oldSystemKey, notNullValue());
+                final String unsigned = service.unsignAndVerify(signed, oldSystemKey);
+                assertThat(unsigned, equalTo(text));
+                final String newSigned = service.sign(unsigned);
+                assertThat(newSigned, not(equalTo(signed)));
+                assertThat(newSigned, not(equalTo(text)));
+                latch.countDown();
+            }
+        });
+
+        // we need to sleep to ensure the timestamp of the file will definitely change
+        // and so the resource watcher will pick up the change.
+        Thread.sleep(1000);
+
+        Files.write(keyFile, InternalCryptoService.generateKey());
+        if (!latch.await(10, TimeUnit.SECONDS)) {
+            fail("waiting too long for test to complete. Expected callback is not called or finished running");
+        }
+    }
+
+    @Test
+    public void testReencryptValuesOnKeyDeleted() throws Exception {
+        final InternalCryptoService service = new InternalCryptoService(settings, env, watcherService).start();
+        assertThat(service.encryptionEnabled(), is(true));
+        final char[] text = randomAsciiOfLength(10).toCharArray();
+        final char[] encrypted = service.encrypt(text);
+        assertThat(text, not(equalTo(encrypted)));
+
+        final CountDownLatch latch = new CountDownLatch(1);
+        service.register(new CryptoService.Listener() {
+            @Override
+            public void onKeyChange(SecretKey oldSystemKey, SecretKey oldEncryptionKey) {
+                final char[] plainText = service.decrypt(encrypted, oldEncryptionKey);
+                assertThat(plainText, equalTo(text));
+                final char[] newEncrypted = service.encrypt(plainText);
+                assertThat(newEncrypted, not(equalTo(encrypted)));
+                assertThat(newEncrypted, equalTo(plainText));
+                latch.countDown();
+            }
+        });
+
+        // we need to sleep to ensure the timestamp of the file will definitely change
+        // and so the resource watcher will pick up the change.
+        Thread.sleep(1000);
+
+        Files.delete(keyFile);
+        if (!latch.await(10, TimeUnit.SECONDS)) {
+            fail("waiting too long for test to complete. Expected callback is not called or finished running");
+        }
+    }
+
+    @Test
+    public void testAllListenersCalledWhenExceptionThrown() throws Exception {
+        final InternalCryptoService service = new InternalCryptoService(settings, env, watcherService).start();
+        assertThat(service.encryptionEnabled(), is(true));
+
+        final CountDownLatch latch = new CountDownLatch(3);
+        service.register(new CryptoService.Listener() {
+            @Override
+            public void onKeyChange(SecretKey oldSystemKey, SecretKey oldEncryptionKey) {
+                latch.countDown();
+            }
+        });
+        service.register(new CryptoService.Listener() {
+            @Override
+            public void onKeyChange(SecretKey oldSystemKey, SecretKey oldEncryptionKey) {
+                latch.countDown();
+                throw new RuntimeException("misbehaving listener");
+            }
+        });
+        service.register(new CryptoService.Listener() {
+            @Override
+            public void onKeyChange(SecretKey oldSystemKey, SecretKey oldEncryptionKey) {
+                latch.countDown();
+            }
+        });
+
+        // we need to sleep to ensure the timestamp of the file will definitely change
+        // and so the resource watcher will pick up the change.
+        Thread.sleep(1000);
+
+        Files.write(keyFile, InternalCryptoService.generateKey());
+        if (!latch.await(10, TimeUnit.SECONDS)) {
+            fail("waiting too long for test to complete. Expected callback is not called or finished running");
+        }
+    }
+
     private static byte[] randomByteArray() {
         return randomByteArray(0);
     }