nifi-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From alopre...@apache.org
Subject [13/13] nifi git commit: NIFI-3594 Implemented encrypted provenance repository. Added src/test/resources/logback-test.xml files resetting log level from DEBUG (in nifi-data-provenance-utils) to WARN because later tests depend on MockComponentLog recordin
Date Tue, 02 May 2017 17:27:38 GMT
NIFI-3594 Implemented encrypted provenance repository.
Added src/test/resources/logback-test.xml files resetting log level from DEBUG (in nifi-data-provenance-utils) to WARN because later tests depend on MockComponentLog recording a certain number of messages and this number is different than expected if the log level is DEBUG.

This closes #1686.

Signed-off-by: Bryan Bende, Yolanda M. Davis, and Mark Payne


Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/7d242076
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/7d242076
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/7d242076

Branch: refs/heads/master
Commit: 7d242076ce1d0a555e2ea68d88f29b7461d37b9a
Parents: fae2e3a
Author: Andy LoPresto <alopresto@apache.org>
Authored: Mon Mar 13 21:53:00 2017 -0700
Committer: Andy LoPresto <alopresto@apache.org>
Committed: Tue May 2 13:24:07 2017 -0400

----------------------------------------------------------------------
 .../nifi/provenance/ProvenanceEventRecord.java  |  15 +-
 nifi-commons/nifi-data-provenance-utils/pom.xml |  16 +
 .../provenance/AESProvenanceEventEncryptor.java | 237 ++++++
 .../org/apache/nifi/provenance/CryptoUtils.java | 248 ++++++
 .../nifi/provenance/EncryptionException.java    |  92 +++
 .../nifi/provenance/EncryptionMetadata.java     |  55 ++
 .../nifi/provenance/FileBasedKeyProvider.java   |  67 ++
 .../org/apache/nifi/provenance/KeyProvider.java |  60 ++
 .../provenance/PlaceholderProvenanceEvent.java  |  10 +
 .../provenance/ProvenanceEventEncryptor.java    |  59 ++
 .../StandardProvenanceEventRecord.java          |  41 +-
 .../nifi/provenance/StaticKeyProvider.java      |  96 +++
 .../AESProvenanceEventEncryptorTest.groovy      | 303 +++++++
 .../nifi/provenance/CryptoUtilsTest.groovy      | 436 ++++++++++
 .../provenance/EncryptionExceptionTest.java     |  27 +
 .../src/test/resources/logback-test.xml         |  32 +
 .../org/apache/nifi/util/NiFiProperties.java    |  74 +-
 nifi-commons/nifi-security-utils/pom.xml        |   4 +
 .../nifi/security/util/EncryptionMethod.java    |  13 +-
 .../util/crypto/AESKeyedCipherProvider.java     | 152 ++++
 .../security/util/crypto/CipherProvider.java    |  23 +
 .../security/util/crypto/CipherUtility.java     | 328 ++++++++
 .../util/crypto/KeyedCipherProvider.java        |  72 ++
 .../AESKeyedCipherProviderGroovyTest.groovy     | 347 ++++++++
 .../util/crypto/CipherUtilityGroovyTest.groovy  | 251 ++++++
 .../apache/nifi/controller/FlowController.java  |  72 +-
 .../properties/ProtectedNiFiProperties.java     |  24 +-
 .../AESSensitivePropertyProviderTest.groovy     |  44 +-
 .../ProtectedNiFiPropertiesGroovyTest.groovy    |  73 +-
 .../StandardNiFiPropertiesGroovyTest.groovy     | 215 ++++-
 ...fi_with_additional_sensitive_keys.properties |   2 +-
 .../nifi-framework/nifi-resources/pom.xml       |   5 +
 .../src/main/resources/conf/nifi.properties     |   5 +
 .../src/test/resources/logback-test.xml         |  32 +
 .../src/test/resources/logback-test.xml         |  32 +
 .../pom.xml                                     |   5 +
 .../provenance/EncryptedSchemaRecordReader.java | 154 ++++
 .../provenance/EncryptedSchemaRecordWriter.java | 199 +++++
 ...EncryptedWriteAheadProvenanceRepository.java | 159 ++++
 .../EventIdFirstSchemaRecordReader.java         |  55 +-
 .../EventIdFirstSchemaRecordWriter.java         |  43 +-
 .../provenance/RepositoryConfiguration.java     |  66 +-
 .../WriteAheadProvenanceRepository.java         |  17 +-
 .../index/lucene/LuceneEventIndex.java          |  17 +-
 .../nifi/provenance/schema/EventFieldNames.java |   8 +
 .../schema/LookupTableEventRecordFields.java    |   6 +-
 .../schema/LookupTableEventSchema.java          |   2 -
 .../serialization/CompressableRecordReader.java |   3 +-
 .../provenance/serialization/RecordReaders.java |  62 +-
 .../provenance/util/StorageSummaryEvent.java    |  11 +-
 ....apache.nifi.provenance.ProvenanceRepository |   3 +-
 ...EncryptedSchemaRecordReaderWriterTest.groovy | 281 +++++++
 ...tedWriteAheadProvenanceRepositoryTest.groovy | 391 +++++++++
 .../AbstractTestRecordReaderWriter.java         |  30 +-
 .../VolatileProvenanceRepository.java           |  65 +-
 .../nifi-standard-processors/pom.xml            |   2 +-
 .../processors/standard/EncryptContent.java     |  31 +-
 .../util/crypto/AESKeyedCipherProvider.java     | 153 ----
 .../util/crypto/BcryptCipherProvider.java       | 177 -----
 .../standard/util/crypto/CipherProvider.java    |  23 -
 .../util/crypto/CipherProviderFactory.java      |  57 --
 .../standard/util/crypto/CipherUtility.java     | 329 --------
 .../util/crypto/KeyedCipherProvider.java        |  73 --
 .../standard/util/crypto/KeyedEncryptor.java    | 163 ----
 .../util/crypto/NiFiLegacyCipherProvider.java   | 136 ----
 .../util/crypto/OpenPGPKeyBasedEncryptor.java   | 380 ---------
 .../crypto/OpenPGPPasswordBasedEncryptor.java   | 158 ----
 .../util/crypto/OpenSSLPKCS5CipherProvider.java | 199 -----
 .../standard/util/crypto/PBECipherProvider.java |  73 --
 .../util/crypto/PBKDF2CipherProvider.java       | 201 -----
 .../util/crypto/PasswordBasedEncryptor.java     | 193 -----
 .../util/crypto/RandomIVPBECipherProvider.java  |  71 --
 .../util/crypto/ScryptCipherProvider.java       | 286 -------
 .../standard/util/crypto/bcrypt/BCrypt.java     | 789 -------------------
 .../standard/util/crypto/scrypt/Scrypt.java     | 511 ------------
 .../util/crypto/BcryptCipherProvider.java       | 176 +++++
 .../util/crypto/CipherProviderFactory.java      |  56 ++
 .../security/util/crypto/KeyedEncryptor.java    | 162 ++++
 .../util/crypto/NiFiLegacyCipherProvider.java   | 135 ++++
 .../util/crypto/OpenPGPKeyBasedEncryptor.java   | 379 +++++++++
 .../crypto/OpenPGPPasswordBasedEncryptor.java   | 157 ++++
 .../util/crypto/OpenSSLPKCS5CipherProvider.java | 198 +++++
 .../security/util/crypto/PBECipherProvider.java |  72 ++
 .../util/crypto/PBKDF2CipherProvider.java       | 200 +++++
 .../util/crypto/PasswordBasedEncryptor.java     | 192 +++++
 .../util/crypto/RandomIVPBECipherProvider.java  |  70 ++
 .../util/crypto/ScryptCipherProvider.java       | 285 +++++++
 .../security/util/crypto/bcrypt/BCrypt.java     | 789 +++++++++++++++++++
 .../security/util/crypto/scrypt/Scrypt.java     | 510 ++++++++++++
 .../standard/TestEncryptContentGroovy.groovy    |   4 +-
 .../AESKeyedCipherProviderGroovyTest.groovy     | 340 --------
 .../BcryptCipherProviderGroovyTest.groovy       | 539 -------------
 .../CipherProviderFactoryGroovyTest.groovy      |  97 ---
 .../util/crypto/CipherUtilityGroovyTest.groovy  | 251 ------
 .../util/crypto/KeyedEncryptorGroovyTest.groovy | 122 ---
 .../NiFiLegacyCipherProviderGroovyTest.groovy   | 294 -------
 .../OpenSSLPKCS5CipherProviderGroovyTest.groovy | 319 --------
 .../PBKDF2CipherProviderGroovyTest.groovy       | 540 -------------
 .../PasswordBasedEncryptorGroovyTest.groovy     | 225 ------
 .../ScryptCipherProviderGroovyTest.groovy       | 597 --------------
 .../util/crypto/scrypt/ScryptGroovyTest.groovy  | 399 ----------
 .../BcryptCipherProviderGroovyTest.groovy       | 538 +++++++++++++
 .../CipherProviderFactoryGroovyTest.groovy      |  97 +++
 .../util/crypto/KeyedEncryptorGroovyTest.groovy | 122 +++
 .../NiFiLegacyCipherProviderGroovyTest.groovy   | 299 +++++++
 .../OpenSSLPKCS5CipherProviderGroovyTest.groovy | 323 ++++++++
 .../PBKDF2CipherProviderGroovyTest.groovy       | 545 +++++++++++++
 .../PasswordBasedEncryptorGroovyTest.groovy     | 225 ++++++
 .../ScryptCipherProviderGroovyTest.groovy       | 597 ++++++++++++++
 .../util/crypto/scrypt/ScryptGroovyTest.groovy  | 399 ++++++++++
 .../processors/standard/TestEncryptContent.java |  15 +-
 .../crypto/OpenPGPKeyBasedEncryptorTest.java    | 132 ----
 .../OpenPGPPasswordBasedEncryptorTest.java      | 123 ---
 .../crypto/OpenPGPKeyBasedEncryptorTest.java    | 131 +++
 .../OpenPGPPasswordBasedEncryptorTest.java      | 122 +++
 .../src/test/resources/logback-test.xml         |   2 +-
 116 files changed, 11711 insertions(+), 8211 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/nifi/blob/7d242076/nifi-api/src/main/java/org/apache/nifi/provenance/ProvenanceEventRecord.java
----------------------------------------------------------------------
diff --git a/nifi-api/src/main/java/org/apache/nifi/provenance/ProvenanceEventRecord.java b/nifi-api/src/main/java/org/apache/nifi/provenance/ProvenanceEventRecord.java
index b05bd85..51bb76f 100644
--- a/nifi-api/src/main/java/org/apache/nifi/provenance/ProvenanceEventRecord.java
+++ b/nifi-api/src/main/java/org/apache/nifi/provenance/ProvenanceEventRecord.java
@@ -188,28 +188,24 @@ public interface ProvenanceEventRecord {
      * {@link ProvenanceEventType#FORK}, {@link ProvenanceEventType#JOIN}, or
      * {@link ProvenanceEventType#CLONE}), or if the queue identifier is
      * unknown, then this method will return <code>null</code>
-     *
      */
     String getSourceQueueIdentifier();
 
     /**
      * @return the Section for the Content Claim that this Event refers to, if
      * any; otherwise, returns <code>null</code>
-     *
      */
     String getContentClaimSection();
 
     /**
      * @return the Section for the Content Claim that the FlowFile previously
      * referenced, if any; otherwise, returns <code>null</code>
-     *
      */
     String getPreviousContentClaimSection();
 
     /**
      * @return the Container for the Content Claim that this Event refers to, if
      * any; otherwise, returns <code>null</code>
-     *
      */
     String getContentClaimContainer();
 
@@ -222,28 +218,31 @@ public interface ProvenanceEventRecord {
     /**
      * @return the Identifier for the Content Claim that this Event refers to,
      * if any; otherwise, returns <code>null</code>
-     *
      */
     String getContentClaimIdentifier();
 
     /**
      * @return the Identifier for the Content Claim that the FlowFile previously
      * referenced, if any; otherwise, returns <code>null</code>
-     *
      */
     String getPreviousContentClaimIdentifier();
 
     /**
      * @return the offset into the Content Claim at which the FlowFile's content
      * begins, if any; otherwise, returns <code>null</code>
-     *
      */
     Long getContentClaimOffset();
 
     /**
      * @return the offset into the Content Claim at which the FlowFile's
      * previous content began, if any; otherwise, returns <code>null</code>
-     *
      */
     Long getPreviousContentClaimOffset();
+
+    /**
+     * Returns the best event identifier for this event (eventId if available, descriptive identifier if not yet persisted to allow for traceability).
+     *
+     * @return a descriptive event ID to allow tracing
+     */
+    String getBestEventIdentifier();
 }

http://git-wip-us.apache.org/repos/asf/nifi/blob/7d242076/nifi-commons/nifi-data-provenance-utils/pom.xml
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-data-provenance-utils/pom.xml b/nifi-commons/nifi-data-provenance-utils/pom.xml
index 6207646..86b25b0 100644
--- a/nifi-commons/nifi-data-provenance-utils/pom.xml
+++ b/nifi-commons/nifi-data-provenance-utils/pom.xml
@@ -34,5 +34,21 @@
             <groupId>org.apache.nifi</groupId>
             <artifactId>nifi-utils</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-security-utils</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.bouncycastle</groupId>
+            <artifactId>bcprov-jdk15on</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-properties-loader</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-properties</artifactId>
+        </dependency>
     </dependencies>
 </project>

http://git-wip-us.apache.org/repos/asf/nifi/blob/7d242076/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/AESProvenanceEventEncryptor.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/AESProvenanceEventEncryptor.java b/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/AESProvenanceEventEncryptor.java
new file mode 100644
index 0000000..d2cc9ca
--- /dev/null
+++ b/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/AESProvenanceEventEncryptor.java
@@ -0,0 +1,237 @@
+/*
+ * 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.nifi.provenance;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.security.KeyManagementException;
+import java.security.SecureRandom;
+import java.security.Security;
+import java.util.Arrays;
+import java.util.List;
+import javax.crypto.BadPaddingException;
+import javax.crypto.Cipher;
+import javax.crypto.IllegalBlockSizeException;
+import javax.crypto.SecretKey;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.security.util.EncryptionMethod;
+import org.apache.nifi.security.util.crypto.AESKeyedCipherProvider;
+import org.bouncycastle.jce.provider.BouncyCastleProvider;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class AESProvenanceEventEncryptor implements ProvenanceEventEncryptor {
+    private static final Logger logger = LoggerFactory.getLogger(AESProvenanceEventEncryptor.class);
+    private static final String ALGORITHM = "AES/GCM/NoPadding";
+    private static final int IV_LENGTH = 16;
+    private static final byte[] EMPTY_IV = new byte[IV_LENGTH];
+    private static final String VERSION = "v1";
+    private static final List<String> SUPPORTED_VERSIONS = Arrays.asList(VERSION);
+    private static final int MIN_METADATA_LENGTH = IV_LENGTH + 3 + 3; // 3 delimiters and 3 non-zero elements
+    private static final int METADATA_DEFAULT_LENGTH = (20 + ALGORITHM.length() + IV_LENGTH + VERSION.length()) * 2; // Default to twice the expected length
+    private static final byte[] SENTINEL = new byte[]{0x01};
+
+    private KeyProvider keyProvider;
+
+    private AESKeyedCipherProvider aesKeyedCipherProvider = new AESKeyedCipherProvider();
+
+    /**
+     * Initializes the encryptor with a {@link KeyProvider}.
+     *
+     * @param keyProvider the key provider which will be responsible for accessing keys
+     * @throws KeyManagementException if there is an issue configuring the key provider
+     */
+    @Override
+    public void initialize(KeyProvider keyProvider) throws KeyManagementException {
+        this.keyProvider = keyProvider;
+
+        if (this.aesKeyedCipherProvider == null) {
+            this.aesKeyedCipherProvider = new AESKeyedCipherProvider();
+        }
+
+        if (Security.getProvider("BC") == null) {
+            Security.addProvider(new BouncyCastleProvider());
+        }
+    }
+
+    /**
+     * Available for dependency injection to override the default {@link AESKeyedCipherProvider} if necessary.
+     *
+     * @param cipherProvider the AES cipher provider to use
+     */
+    void setCipherProvider(AESKeyedCipherProvider cipherProvider) {
+        this.aesKeyedCipherProvider = cipherProvider;
+    }
+
+    /**
+     * Encrypts the provided {@link ProvenanceEventRecord}, serialized to a byte[] by the RecordWriter.
+     *
+     * @param plainRecord the plain record, serialized to a byte[]
+     * @param recordId    an identifier for this record (eventId, generated, etc.)
+     * @param keyId       the ID of the key to use
+     * @return the encrypted record
+     * @throws EncryptionException if there is an issue encrypting this record
+     */
+    @Override
+    public byte[] encrypt(byte[] plainRecord, String recordId, String keyId) throws EncryptionException {
+        if (plainRecord == null || CryptoUtils.isEmpty(keyId)) {
+            throw new EncryptionException("The provenance record and key ID cannot be missing");
+        }
+
+        if (keyProvider == null || !keyProvider.keyExists(keyId)) {
+            throw new EncryptionException("The requested key ID is not available");
+        } else {
+            byte[] ivBytes = new byte[IV_LENGTH];
+            new SecureRandom().nextBytes(ivBytes);
+            try {
+                logger.debug("Encrypting provenance record " + recordId + " with key ID " + keyId);
+                Cipher cipher = initCipher(EncryptionMethod.AES_GCM, Cipher.ENCRYPT_MODE, keyProvider.getKey(keyId), ivBytes);
+                ivBytes = cipher.getIV();
+
+                // Perform the actual encryption
+                byte[] cipherBytes = cipher.doFinal(plainRecord);
+
+                // Serialize and concat encryption details fields (keyId, algo, IV, version, CB length) outside of encryption
+                EncryptionMetadata metadata = new EncryptionMetadata(keyId, ALGORITHM, ivBytes, VERSION, cipherBytes.length);
+                byte[] serializedEncryptionMetadata = serializeEncryptionMetadata(metadata);
+
+                // Add the sentinel byte of 0x01
+                logger.debug("Encrypted provenance event record " + recordId + " with key ID " + keyId);
+                return CryptoUtils.concatByteArrays(SENTINEL, serializedEncryptionMetadata, cipherBytes);
+            } catch (EncryptionException | BadPaddingException | IllegalBlockSizeException | IOException | KeyManagementException e) {
+                final String msg = "Encountered an exception encrypting provenance record " + recordId;
+                logger.error(msg, e);
+                throw new EncryptionException(msg, e);
+            }
+        }
+    }
+
+    private byte[] serializeEncryptionMetadata(EncryptionMetadata metadata) throws IOException {
+        final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        ObjectOutputStream outputStream = new ObjectOutputStream(baos);
+        outputStream.writeObject(metadata);
+        outputStream.close();
+        return baos.toByteArray();
+    }
+
+    private Cipher initCipher(EncryptionMethod method, int mode, SecretKey key, byte[] ivBytes) throws EncryptionException {
+        try {
+            if (method == null || key == null || ivBytes == null) {
+                throw new IllegalArgumentException("Missing critical information");
+            }
+            return aesKeyedCipherProvider.getCipher(method, key, ivBytes, mode == Cipher.ENCRYPT_MODE);
+        } catch (Exception e) {
+            logger.error("Encountered an exception initializing the cipher", e);
+            throw new EncryptionException(e);
+        }
+    }
+
+    /**
+     * Decrypts the provided byte[] (an encrypted record with accompanying metadata).
+     *
+     * @param encryptedRecord the encrypted record in byte[] form
+     * @param recordId        an identifier for this record (eventId, generated, etc.)
+     * @return the decrypted record
+     * @throws EncryptionException if there is an issue decrypting this record
+     */
+    @Override
+    public byte[] decrypt(byte[] encryptedRecord, String recordId) throws EncryptionException {
+        if (encryptedRecord == null) {
+            throw new EncryptionException("The encrypted provenance record cannot be missing");
+        }
+
+        EncryptionMetadata metadata;
+        try {
+            metadata = extractEncryptionMetadata(encryptedRecord);
+        } catch (IOException | ClassNotFoundException e) {
+            final String msg = "Encountered an error reading the encryption metadata: ";
+            logger.error(msg, e);
+            throw new EncryptionException(msg, e);
+        }
+
+        if (!SUPPORTED_VERSIONS.contains(metadata.version)) {
+            throw new EncryptionException("The event was encrypted with version " + metadata.version + " which is not in the list of supported versions " + StringUtils.join(SUPPORTED_VERSIONS, ","));
+        }
+
+        // TODO: Actually use the version to determine schema, etc.
+
+        if (keyProvider == null || !keyProvider.keyExists(metadata.keyId) || CryptoUtils.isEmpty(metadata.keyId)) {
+            throw new EncryptionException("The requested key ID " + metadata.keyId + " is not available");
+        } else {
+            try {
+                logger.debug("Decrypting provenance record " + recordId + " with key ID " + metadata.keyId);
+                EncryptionMethod method = EncryptionMethod.forAlgorithm(metadata.algorithm);
+                Cipher cipher = initCipher(method, Cipher.DECRYPT_MODE, keyProvider.getKey(metadata.keyId), metadata.ivBytes);
+
+                // Strip the metadata away to get just the cipher bytes
+                byte[] cipherBytes = extractCipherBytes(encryptedRecord, metadata);
+
+                // Perform the actual decryption
+                byte[] plainBytes = cipher.doFinal(cipherBytes);
+
+                logger.debug("Decrypted provenance event record " + recordId + " with key ID " + metadata.keyId);
+                return plainBytes;
+            } catch (EncryptionException | BadPaddingException | IllegalBlockSizeException | KeyManagementException e) {
+                final String msg = "Encountered an exception decrypting provenance record " + recordId;
+                logger.error(msg, e);
+                throw new EncryptionException(msg, e);
+            }
+        }
+    }
+
+    /**
+     * Returns a valid key identifier for this encryptor (valid for encryption and decryption) or throws an exception if none are available.
+     *
+     * @return the key ID
+     * @throws KeyManagementException if no available key IDs are valid for both operations
+     */
+    @Override
+    public String getNextKeyId() throws KeyManagementException {
+        if (keyProvider != null) {
+            List<String> availableKeyIds = keyProvider.getAvailableKeyIds();
+            if (!availableKeyIds.isEmpty()) {
+                return availableKeyIds.get(0);
+            }
+        }
+        throw new KeyManagementException("No available key IDs");
+    }
+
+    private EncryptionMetadata extractEncryptionMetadata(byte[] encryptedRecord) throws EncryptionException, IOException, ClassNotFoundException {
+        if (encryptedRecord == null || encryptedRecord.length < MIN_METADATA_LENGTH) {
+            throw new EncryptionException("The encrypted record is too short to contain the metadata");
+        }
+
+        // Skip the first byte (SENTINEL) and don't need to copy all the serialized record
+        ByteArrayInputStream bais = new ByteArrayInputStream(encryptedRecord);
+        bais.read();
+        try (ObjectInputStream ois = new ObjectInputStream(bais)) {
+            return (EncryptionMetadata) ois.readObject();
+        }
+    }
+
+    private byte[] extractCipherBytes(byte[] encryptedRecord, EncryptionMetadata metadata) {
+        return Arrays.copyOfRange(encryptedRecord, encryptedRecord.length - metadata.cipherByteLength, encryptedRecord.length);
+    }
+
+    @Override
+    public String toString() {
+        return "AES Provenance Event Encryptor with Key Provider: " + keyProvider.toString();
+    }
+}

http://git-wip-us.apache.org/repos/asf/nifi/blob/7d242076/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/CryptoUtils.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/CryptoUtils.java b/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/CryptoUtils.java
new file mode 100644
index 0000000..1b8f11a
--- /dev/null
+++ b/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/CryptoUtils.java
@@ -0,0 +1,248 @@
+/*
+ * 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.nifi.provenance;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileReader;
+import java.io.IOException;
+import java.security.KeyManagementException;
+import java.security.NoSuchAlgorithmException;
+import java.util.Arrays;
+import java.util.Base64;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Pattern;
+import javax.crypto.BadPaddingException;
+import javax.crypto.Cipher;
+import javax.crypto.IllegalBlockSizeException;
+import javax.crypto.SecretKey;
+import javax.crypto.spec.SecretKeySpec;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.security.util.EncryptionMethod;
+import org.apache.nifi.security.util.crypto.AESKeyedCipherProvider;
+import org.apache.nifi.util.NiFiProperties;
+import org.bouncycastle.util.encoders.Hex;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CryptoUtils {
+    private static final Logger logger = LoggerFactory.getLogger(StaticKeyProvider.class);
+    private static final String STATIC_KEY_PROVIDER_CLASS_NAME = "org.apache.nifi.provenance.StaticKeyProvider";
+    private static final String FILE_BASED_KEY_PROVIDER_CLASS_NAME = "org.apache.nifi.provenance.FileBasedKeyProvider";
+    private static final Pattern HEX_PATTERN = Pattern.compile("(?i)^[0-9a-f]+$");
+
+    private static final List<Integer> UNLIMITED_KEY_LENGTHS = Arrays.asList(32, 48, 64);
+
+    public static final int IV_LENGTH = 16;
+
+    public static boolean isUnlimitedStrengthCryptoAvailable() {
+        try {
+            return Cipher.getMaxAllowedKeyLength("AES") > 128;
+        } catch (NoSuchAlgorithmException e) {
+            logger.warn("Tried to determine if unlimited strength crypto is available but the AES algorithm is not available");
+            return false;
+        }
+    }
+
+    /**
+     * Utility method which returns true if the string is null, empty, or entirely whitespace.
+     *
+     * @param src the string to evaluate
+     * @return true if empty
+     */
+    public static boolean isEmpty(String src) {
+        return src == null || src.trim().isEmpty();
+    }
+
+    /**
+     * Concatenates multiple byte[] into a single byte[].
+     *
+     * @param arrays the component byte[] in order
+     * @return a concatenated byte[]
+     * @throws IOException this should never be thrown
+     */
+    public static byte[] concatByteArrays(byte[]... arrays) throws IOException {
+        int totalByteLength = 0;
+        for (byte[] bytes : arrays) {
+            totalByteLength += bytes.length;
+        }
+        byte[] totalBytes = new byte[totalByteLength];
+        int currentLength = 0;
+        for (byte[] bytes : arrays) {
+            System.arraycopy(bytes, 0, totalBytes, currentLength, bytes.length);
+            currentLength += bytes.length;
+        }
+        return totalBytes;
+    }
+
+    /**
+     * Returns true if the provided configuration values successfully define the specified {@link KeyProvider}.
+     *
+     * @param keyProviderImplementation the FQ class name of the {@link KeyProvider} implementation
+     * @param keyProviderLocation       the location of the definition (for {@link FileBasedKeyProvider}, etc.)
+     * @param keyId                     the active key ID
+     * @param encryptionKeys            a map of key IDs to key material in hex format
+     * @return true if the provided configuration is valid
+     */
+    public static boolean isValidKeyProvider(String keyProviderImplementation, String keyProviderLocation, String keyId, Map<String, String> encryptionKeys) {
+        if (STATIC_KEY_PROVIDER_CLASS_NAME.equals(keyProviderImplementation)) {
+            // Ensure the keyId and key(s) are valid
+            if (encryptionKeys == null) {
+                return false;
+            } else {
+                boolean everyKeyValid = encryptionKeys.values().stream().allMatch(CryptoUtils::keyIsValid);
+                return everyKeyValid && StringUtils.isNotEmpty(keyId);
+            }
+        } else if (FILE_BASED_KEY_PROVIDER_CLASS_NAME.equals(keyProviderImplementation)) {
+            // Ensure the file can be read and the keyId is populated (does not read file to validate)
+            final File kpf = new File(keyProviderLocation);
+            return kpf.exists() && kpf.canRead() && StringUtils.isNotEmpty(keyId);
+        } else {
+            logger.error("The attempt to validate the key provider failed keyProviderImplementation = "
+                    + keyProviderImplementation + " , keyProviderLocation = "
+                    + keyProviderLocation + " , keyId = "
+                    + keyId + " , encryptionKeys = "
+                    + ((encryptionKeys == null) ? "0" : encryptionKeys.size()));
+
+            return false;
+        }
+    }
+
+    /**
+     * Returns true if the provided key is valid hex and is the correct length for the current system's JCE policies.
+     *
+     * @param encryptionKeyHex the key in hexadecimal
+     * @return true if this key is valid
+     */
+    public static boolean keyIsValid(String encryptionKeyHex) {
+        return isHexString(encryptionKeyHex)
+                && (isUnlimitedStrengthCryptoAvailable()
+                ? UNLIMITED_KEY_LENGTHS.contains(encryptionKeyHex.length())
+                : encryptionKeyHex.length() == 32);
+    }
+
+    /**
+     * Returns true if the input is valid hexadecimal (does not enforce length and is case-insensitive).
+     *
+     * @param hexString the string to evaluate
+     * @return true if the string is valid hex
+     */
+    public static boolean isHexString(String hexString) {
+        return StringUtils.isNotEmpty(hexString) && HEX_PATTERN.matcher(hexString).matches();
+    }
+
+    /**
+     * Returns a {@link SecretKey} formed from the hexadecimal key bytes (validity is checked).
+     *
+     * @param keyHex the key in hex form
+     * @return the SecretKey
+     */
+    public static SecretKey formKeyFromHex(String keyHex) throws KeyManagementException {
+        if (keyIsValid(keyHex)) {
+            return new SecretKeySpec(Hex.decode(keyHex), "AES");
+        } else {
+            throw new KeyManagementException("The provided key material is not valid");
+        }
+    }
+
+    /**
+     * Returns a map containing the key IDs and the parsed key from a key provider definition file.
+     * The values in the file are decrypted using the master key provided. If the file is missing or empty,
+     * cannot be read, or if no valid keys are read, a {@link KeyManagementException} will be thrown.
+     *
+     * @param filepath  the key definition file path
+     * @param masterKey the master key used to decrypt each key definition
+     * @return a Map of key IDs to SecretKeys
+     * @throws KeyManagementException if the file is missing or invalid
+     */
+    public static Map<String, SecretKey> readKeys(String filepath, SecretKey masterKey) throws KeyManagementException {
+        Map<String, SecretKey> keys = new HashMap<>();
+
+        if (StringUtils.isBlank(filepath)) {
+            throw new KeyManagementException("The key provider file is not present and readable");
+        }
+        File file = new File(filepath);
+        if (!file.exists() || !file.canRead()) {
+            throw new KeyManagementException("The key provider file is not present and readable");
+        }
+
+        try (BufferedReader br = new BufferedReader(new FileReader(file))) {
+            AESKeyedCipherProvider masterCipherProvider = new AESKeyedCipherProvider();
+
+            String line;
+            int l = 1;
+            while ((line = br.readLine()) != null) {
+                String[] components = line.split("=", 2);
+                if (components.length != 2 || StringUtils.isAnyEmpty(components)) {
+                    logger.warn("Line " + l + " is not properly formatted -- keyId=Base64EncodedKey...");
+                }
+                String keyId = components[0];
+                if (StringUtils.isNotEmpty(keyId)) {
+                    try {
+                        byte[] base64Bytes = Base64.getDecoder().decode(components[1]);
+                        byte[] ivBytes = Arrays.copyOfRange(base64Bytes, 0, IV_LENGTH);
+
+                        Cipher masterCipher = null;
+                        try {
+                            masterCipher = masterCipherProvider.getCipher(EncryptionMethod.AES_GCM, masterKey, ivBytes, false);
+                        } catch (Exception e) {
+                            throw new KeyManagementException("Error building cipher to decrypt FileBaseKeyProvider definition at " + filepath, e);
+                        }
+                        byte[] individualKeyBytes = masterCipher.doFinal(Arrays.copyOfRange(base64Bytes, IV_LENGTH, base64Bytes.length));
+
+                        SecretKey key = new SecretKeySpec(individualKeyBytes, "AES");
+                        logger.debug("Read and decrypted key for " + keyId);
+                        if (keys.containsKey(keyId)) {
+                            logger.warn("Multiple key values defined for " + keyId + " -- using most recent value");
+                        }
+                        keys.put(keyId, key);
+                    } catch (IllegalArgumentException e) {
+                        logger.error("Encountered an error decoding Base64 for " + keyId + ": " + e.getLocalizedMessage());
+                    } catch (BadPaddingException | IllegalBlockSizeException e) {
+                        logger.error("Encountered an error decrypting key for " + keyId + ": " + e.getLocalizedMessage());
+                    }
+                }
+                l++;
+            }
+
+            if (keys.isEmpty()) {
+                throw new KeyManagementException("The provided file contained no valid keys");
+            }
+
+            logger.info("Read " + keys.size() + " keys from FileBasedKeyProvider " + filepath);
+            return keys;
+        } catch (IOException e) {
+            throw new KeyManagementException("Error reading FileBasedKeyProvider definition at " + filepath, e);
+        }
+
+    }
+
+    public static boolean isProvenanceRepositoryEncryptionConfigured(NiFiProperties niFiProperties) {
+        final String implementationClassName = niFiProperties.getProperty(NiFiProperties.PROVENANCE_REPO_IMPLEMENTATION_CLASS);
+        // Referencing EWAPR.class.getName() would require a dependency on the module
+        boolean encryptedRepo = "org.apache.nifi.provenance.EncryptedWriteAheadProvenanceRepository".equals(implementationClassName);
+        boolean keyProviderConfigured = isValidKeyProvider(
+                niFiProperties.getProperty(NiFiProperties.PROVENANCE_REPO_ENCRYPTION_KEY_PROVIDER_IMPLEMENTATION_CLASS),
+                niFiProperties.getProperty(NiFiProperties.PROVENANCE_REPO_ENCRYPTION_KEY_PROVIDER_LOCATION),
+                niFiProperties.getProvenanceRepoEncryptionKeyId(),
+                niFiProperties.getProvenanceRepoEncryptionKeys());
+
+        return encryptedRepo && keyProviderConfigured;
+    }
+}

http://git-wip-us.apache.org/repos/asf/nifi/blob/7d242076/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/EncryptionException.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/EncryptionException.java b/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/EncryptionException.java
new file mode 100644
index 0000000..05c52e5
--- /dev/null
+++ b/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/EncryptionException.java
@@ -0,0 +1,92 @@
+/*
+ * 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.nifi.provenance;
+
+public class EncryptionException extends Throwable {
+    /**
+     * Constructs a new EncryptionException with {@code null} as its detail message.
+     * The cause is not initialized, and may subsequently be initialized by a
+     * call to {@link #initCause}.
+     * <p>
+     * <p>The {@link #fillInStackTrace()} method is called to initialize
+     * the stack trace data in the newly created exception.
+     */
+    public EncryptionException() {
+        super();
+    }
+
+    /**
+     * Constructs a new EncryptionException with the specified detail message.  The
+     * cause is not initialized, and may subsequently be initialized by
+     * a call to {@link #initCause}.
+     * <p>
+     * <p>The {@link #fillInStackTrace()} method is called to initialize
+     * the stack trace data in the newly created exception.
+     *
+     * @param message the detail message. The detail message is saved for
+     *                later retrieval by the {@link #getMessage()} method.
+     */
+    public EncryptionException(String message) {
+        super(message);
+    }
+
+    /**
+     * Constructs a new EncryptionException with the specified detail message and
+     * cause.  <p>Note that the detail message associated with
+     * {@code cause} is <i>not</i> automatically incorporated in
+     * this exception's detail message.
+     * <p>
+     * <p>The {@link #fillInStackTrace()} method is called to initialize
+     * the stack trace data in the newly created throwable.
+     *
+     * @param message the detail message (which is saved for later retrieval
+     *                by the {@link #getMessage()} method).
+     * @param cause   the cause (which is saved for later retrieval by the
+     *                {@link #getCause()} method).  (A {@code null} value is
+     *                permitted, and indicates that the cause is nonexistent or
+     *                unknown.)
+     * @since 1.4
+     */
+    public EncryptionException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    /**
+     * Constructs a new EncryptionException with the specified cause and a detail
+     * message of {@code (cause==null ? null : cause.toString())} (which
+     * typically contains the class and detail message of {@code cause}).
+     * This constructor is useful for exceptions that are little more than
+     * wrappers for other exceptions.
+     * <p>
+     * <p>The {@link #fillInStackTrace()} method is called to initialize
+     * the stack trace data in the newly created exception.
+     *
+     * @param cause the cause (which is saved for later retrieval by the
+     *              {@link #getCause()} method).  (A {@code null} value is
+     *              permitted, and indicates that the cause is nonexistent or
+     *              unknown.)
+     * @since 1.4
+     */
+    public EncryptionException(Throwable cause) {
+        super(cause);
+    }
+
+    @Override
+    public String toString() {
+        return "EncryptionException " + getLocalizedMessage();
+    }
+}

http://git-wip-us.apache.org/repos/asf/nifi/blob/7d242076/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/EncryptionMetadata.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/EncryptionMetadata.java b/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/EncryptionMetadata.java
new file mode 100644
index 0000000..0d969da
--- /dev/null
+++ b/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/EncryptionMetadata.java
@@ -0,0 +1,55 @@
+/*
+ * 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.nifi.provenance;
+
+import java.io.Serializable;
+import org.apache.commons.codec.binary.Hex;
+
+public class EncryptionMetadata implements Serializable {
+    protected String keyId;
+    protected String algorithm;
+    protected byte[] ivBytes;
+    protected String version;
+    protected int cipherByteLength;
+
+    EncryptionMetadata() {
+    }
+
+    EncryptionMetadata(String keyId, String algorithm, byte[] ivBytes, String version, int cipherByteLength) {
+        this.keyId = keyId;
+        this.ivBytes = ivBytes;
+        this.algorithm = algorithm;
+        this.version = version;
+        this.cipherByteLength = cipherByteLength;
+    }
+
+    @Override
+    public String toString() {
+        String sb = "AES Provenance Record Encryption Metadata" +
+                " Key ID: " +
+                keyId +
+                " Algorithm: " +
+                algorithm +
+                " IV: " +
+                Hex.encodeHexString(ivBytes) +
+                " Version: " +
+                version +
+                " Cipher text length: " +
+                cipherByteLength;
+        return sb;
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/nifi/blob/7d242076/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/FileBasedKeyProvider.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/FileBasedKeyProvider.java b/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/FileBasedKeyProvider.java
new file mode 100644
index 0000000..b70b3e8
--- /dev/null
+++ b/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/FileBasedKeyProvider.java
@@ -0,0 +1,67 @@
+/*
+ * 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.nifi.provenance;
+
+import java.io.IOException;
+import java.security.KeyManagementException;
+import javax.crypto.SecretKey;
+import javax.crypto.spec.SecretKeySpec;
+import javax.naming.OperationNotSupportedException;
+import org.apache.nifi.properties.NiFiPropertiesLoader;
+import org.bouncycastle.util.encoders.Hex;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class FileBasedKeyProvider extends StaticKeyProvider {
+    private static final Logger logger = LoggerFactory.getLogger(FileBasedKeyProvider.class);
+
+    private String filepath;
+
+    FileBasedKeyProvider(String location) throws KeyManagementException {
+        this(location, getMasterKey());
+    }
+
+    FileBasedKeyProvider(String location, SecretKey masterKey) throws KeyManagementException {
+        super(CryptoUtils.readKeys(location, masterKey));
+        this.filepath = location;
+    }
+
+    private static SecretKey getMasterKey() throws KeyManagementException {
+        try {
+            // Get the master encryption key from bootstrap.conf
+            String masterKeyHex = NiFiPropertiesLoader.extractKeyFromBootstrapFile();
+            return new SecretKeySpec(Hex.decode(masterKeyHex), "AES");
+        } catch (IOException e) {
+            logger.error("Encountered an error: ", e);
+            throw new KeyManagementException(e);
+        }
+    }
+
+    /**
+     * Adds the key to the provider and associates it with the given ID. Some implementations may not allow this operation.
+     *
+     * @param keyId the key identifier
+     * @param key   the key
+     * @return true if the key was successfully added
+     * @throws OperationNotSupportedException if this implementation doesn't support adding keys
+     * @throws KeyManagementException         if the key is invalid, the ID conflicts, etc.
+     */
+    @Override
+    public boolean addKey(String keyId, SecretKey key) throws OperationNotSupportedException, KeyManagementException {
+        throw new OperationNotSupportedException("This implementation does not allow adding keys. Modify the file backing this provider at " + filepath);
+    }
+}

http://git-wip-us.apache.org/repos/asf/nifi/blob/7d242076/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/KeyProvider.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/KeyProvider.java b/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/KeyProvider.java
new file mode 100644
index 0000000..39f6384
--- /dev/null
+++ b/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/KeyProvider.java
@@ -0,0 +1,60 @@
+/*
+ * 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.nifi.provenance;
+
+import java.security.KeyManagementException;
+import java.util.List;
+import javax.crypto.SecretKey;
+import javax.naming.OperationNotSupportedException;
+
+public interface KeyProvider {
+
+    /**
+     * Returns the key identified by this ID or throws an exception if one is not available.
+     *
+     * @param keyId the key identifier
+     * @return the key
+     * @throws KeyManagementException if the key cannot be retrieved
+     */
+     SecretKey getKey(String keyId) throws KeyManagementException;
+
+    /**
+     * Returns true if the key exists and is available. Null or empty IDs will return false.
+     *
+     * @param keyId the key identifier
+     * @return true if the key can be used
+     */
+     boolean keyExists(String keyId);
+
+    /**
+     * Returns a list of available key identifiers (useful for encryption, as retired keys may not be listed here even if they are available for decryption for legacy/BC reasons).
+     *
+     * @return a List of keyIds (empty list if none are available)
+     */
+     List<String> getAvailableKeyIds();
+
+    /**
+     * Adds the key to the provider and associates it with the given ID. Some implementations may not allow this operation.
+     *
+     * @param keyId the key identifier
+     * @param key the key
+     * @return true if the key was successfully added
+     * @throws OperationNotSupportedException if this implementation doesn't support adding keys
+     * @throws KeyManagementException if the key is invalid, the ID conflicts, etc.
+     */
+     boolean addKey(String keyId, SecretKey key) throws OperationNotSupportedException, KeyManagementException;
+}

http://git-wip-us.apache.org/repos/asf/nifi/blob/7d242076/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/PlaceholderProvenanceEvent.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/PlaceholderProvenanceEvent.java b/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/PlaceholderProvenanceEvent.java
index 26696c8..5644355 100644
--- a/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/PlaceholderProvenanceEvent.java
+++ b/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/PlaceholderProvenanceEvent.java
@@ -187,4 +187,14 @@ public class PlaceholderProvenanceEvent implements ProvenanceEventRecord {
     public Long getPreviousContentClaimOffset() {
         return null;
     }
+
+    /**
+     * Returns the best event identifier for this event (eventId if available, descriptive identifier if not yet persisted to allow for traceability).
+     *
+     * @return a descriptive event ID to allow tracing
+     */
+    @Override
+    public String getBestEventIdentifier() {
+        return Long.toString(getEventId());
+    }
 }

http://git-wip-us.apache.org/repos/asf/nifi/blob/7d242076/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/ProvenanceEventEncryptor.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/ProvenanceEventEncryptor.java b/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/ProvenanceEventEncryptor.java
new file mode 100644
index 0000000..c7690e1
--- /dev/null
+++ b/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/ProvenanceEventEncryptor.java
@@ -0,0 +1,59 @@
+/*
+ * 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.nifi.provenance;
+
+import java.security.KeyManagementException;
+
+public interface ProvenanceEventEncryptor {
+
+    /**
+     * Initializes the encryptor with a {@link KeyProvider}.
+     *
+     * @param keyProvider the key provider which will be responsible for accessing keys
+     * @throws KeyManagementException if there is an issue configuring the key provider
+     */
+    void initialize(KeyProvider keyProvider) throws KeyManagementException;
+
+    /**
+     * Encrypts the provided {@link ProvenanceEventRecord}, serialized to a byte[] by the RecordWriter.
+     *
+     * @param plainRecord the plain record, serialized to a byte[]
+     * @param recordId    an identifier for this record (eventId, generated, etc.)
+     * @param keyId       the ID of the key to use
+     * @return the encrypted record
+     * @throws EncryptionException if there is an issue encrypting this record
+     */
+    byte[] encrypt(byte[] plainRecord, String recordId, String keyId) throws EncryptionException;
+
+    /**
+     * Decrypts the provided byte[] (an encrypted record with accompanying metadata).
+     *
+     * @param encryptedRecord the encrypted record in byte[] form
+     * @param recordId        an identifier for this record (eventId, generated, etc.)
+     * @return the decrypted record
+     * @throws EncryptionException if there is an issue decrypting this record
+     */
+    byte[] decrypt(byte[] encryptedRecord, String recordId) throws EncryptionException;
+
+    /**
+     * Returns a valid key identifier for this encryptor (valid for encryption and decryption) or throws an exception if none are available.
+     *
+     * @return the key ID
+     * @throws KeyManagementException if no available key IDs are valid for both operations
+     */
+    String getNextKeyId() throws KeyManagementException;
+}

http://git-wip-us.apache.org/repos/asf/nifi/blob/7d242076/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/StandardProvenanceEventRecord.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/StandardProvenanceEventRecord.java b/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/StandardProvenanceEventRecord.java
index ac60d4f..84e7419 100644
--- a/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/StandardProvenanceEventRecord.java
+++ b/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/StandardProvenanceEventRecord.java
@@ -22,7 +22,6 @@ import java.util.Date;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-
 import org.apache.nifi.flowfile.FlowFile;
 import org.apache.nifi.flowfile.attributes.CoreAttributes;
 import org.apache.nifi.processor.Relationship;
@@ -30,7 +29,7 @@ import org.apache.nifi.processor.Relationship;
 /**
  * Holder for provenance relevant information
  */
-public final class StandardProvenanceEventRecord implements ProvenanceEventRecord {
+public class StandardProvenanceEventRecord implements ProvenanceEventRecord {
 
     private final long eventTime;
     private final long entryDate;
@@ -69,7 +68,7 @@ public final class StandardProvenanceEventRecord implements ProvenanceEventRecor
 
     private volatile long eventId = -1L;
 
-    private StandardProvenanceEventRecord(final Builder builder) {
+    StandardProvenanceEventRecord(final Builder builder) {
         this.eventTime = builder.eventTime;
         this.entryDate = builder.entryDate;
         this.eventType = builder.eventType;
@@ -100,8 +99,8 @@ public final class StandardProvenanceEventRecord implements ProvenanceEventRecor
         contentClaimOffset = builder.contentClaimOffset;
         contentSize = builder.contentSize;
 
-        previousAttributes = builder.previousAttributes == null ? Collections.<String, String>emptyMap() : Collections.unmodifiableMap(builder.previousAttributes);
-        updatedAttributes = builder.updatedAttributes == null ? Collections.<String, String>emptyMap() : Collections.unmodifiableMap(builder.updatedAttributes);
+        previousAttributes = builder.previousAttributes == null ? Collections.emptyMap() : Collections.unmodifiableMap(builder.previousAttributes);
+        updatedAttributes = builder.updatedAttributes == null ? Collections.emptyMap() : Collections.unmodifiableMap(builder.updatedAttributes);
 
         sourceQueueIdentifier = builder.sourceQueueIdentifier;
 
@@ -110,6 +109,11 @@ public final class StandardProvenanceEventRecord implements ProvenanceEventRecor
         }
     }
 
+    public static StandardProvenanceEventRecord copy(StandardProvenanceEventRecord other) {
+        Builder builder = new Builder().fromEvent(other);
+        return new StandardProvenanceEventRecord(builder);
+    }
+
     public String getStorageFilename() {
         return storageFilename;
     }
@@ -199,12 +203,12 @@ public final class StandardProvenanceEventRecord implements ProvenanceEventRecor
 
     @Override
     public List<String> getParentUuids() {
-        return parentUuids == null ? Collections.<String>emptyList() : parentUuids;
+        return parentUuids == null ? Collections.emptyList() : parentUuids;
     }
 
     @Override
     public List<String> getChildUuids() {
-        return childrenUuids == null ? Collections.<String>emptyList() : childrenUuids;
+        return childrenUuids == null ? Collections.emptyList() : childrenUuids;
     }
 
     @Override
@@ -299,8 +303,8 @@ public final class StandardProvenanceEventRecord implements ProvenanceEventRecor
         }
 
         return -37423 + 3 * componentId.hashCode() + (transitUri == null ? 0 : 41 * transitUri.hashCode())
-            + (relationship == null ? 0 : 47 * relationship.hashCode()) + 44 * eventTypeCode
-            + 47 * getChildUuids().hashCode() + 47 * getParentUuids().hashCode();
+                + (relationship == null ? 0 : 47 * relationship.hashCode()) + 44 * eventTypeCode
+                + 47 * getChildUuids().hashCode() + 47 * getParentUuids().hashCode();
     }
 
     @Override
@@ -419,6 +423,23 @@ public final class StandardProvenanceEventRecord implements ProvenanceEventRecor
                 + ", alternateIdentifierUri=" + alternateIdentifierUri + "]";
     }
 
+    /**
+     * Returns a unique identifier for the record. By default, it uses
+     * {@link ProvenanceEventRecord#getEventId()} but if it has not been persisted to the
+     * repository, this is {@code -1}, so it constructs a String of the format
+     * {@code <event type>_on_<flowfile UUID>_by_<component UUID>_at_<event time>}.
+     *
+     * @return a String identifying the record for later analysis
+     */
+    @Override
+    public String getBestEventIdentifier() {
+        if (getEventId() != -1) {
+            return Long.toString(getEventId());
+        } else {
+            return getEventType().name() + "_on_" + getFlowFileUuid() + "_by_" + getComponentId() + "_at_" + getEventTime();
+        }
+    }
+
     public static class Builder implements ProvenanceEventBuilder {
 
         private long eventTime = System.currentTimeMillis();
@@ -733,7 +754,7 @@ public final class StandardProvenanceEventRecord implements ProvenanceEventRecor
         public ProvenanceEventBuilder fromFlowFile(final FlowFile flowFile) {
             setFlowFileEntryDate(flowFile.getEntryDate());
             setLineageStartDate(flowFile.getLineageStartDate());
-            setAttributes(Collections.<String, String>emptyMap(), flowFile.getAttributes());
+            setAttributes(Collections.emptyMap(), flowFile.getAttributes());
             uuid = flowFile.getAttribute(CoreAttributes.UUID.key());
             this.contentSize = flowFile.getSize();
             return this;

http://git-wip-us.apache.org/repos/asf/nifi/blob/7d242076/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/StaticKeyProvider.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/StaticKeyProvider.java b/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/StaticKeyProvider.java
new file mode 100644
index 0000000..e0981dc
--- /dev/null
+++ b/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/StaticKeyProvider.java
@@ -0,0 +1,96 @@
+/*
+ * 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.nifi.provenance;
+
+import java.security.KeyManagementException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import javax.crypto.SecretKey;
+import javax.naming.OperationNotSupportedException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Reference implementation for static key provider (used during tests).
+ */
+public class StaticKeyProvider implements KeyProvider {
+    private static final Logger logger = LoggerFactory.getLogger(StaticKeyProvider.class);
+
+    private Map<String, SecretKey> keys = new HashMap<>();
+
+    StaticKeyProvider(String keyId, String keyHex) throws KeyManagementException {
+        this.keys.put(keyId, CryptoUtils.formKeyFromHex(keyHex));
+    }
+
+    StaticKeyProvider(Map<String, SecretKey> keys) throws KeyManagementException {
+        this.keys.putAll(keys);
+    }
+
+    /**
+     * Returns the key identified by this ID or throws an exception if one is not available.
+     *
+     * @param keyId the key identifier
+     * @return the key
+     * @throws KeyManagementException if the key cannot be retrieved
+     */
+    @Override
+    public SecretKey getKey(String keyId) throws KeyManagementException {
+        logger.debug("Attempting to get key: " + keyId);
+        if (keyExists(keyId)) {
+            return keys.get(keyId);
+        } else {
+            throw new KeyManagementException("No key available for " + keyId);
+        }
+    }
+
+    /**
+     * Returns true if the key exists and is available. Null or empty IDs will return false.
+     *
+     * @param keyId the key identifier
+     * @return true if the key can be used
+     */
+    @Override
+    public boolean keyExists(String keyId) {
+        return keys.containsKey(keyId);
+    }
+
+    /**
+     * Returns a singleton list of the available key identifier.
+     *
+     * @return a List containing the {@code KEY_ID}
+     */
+    @Override
+    public List<String> getAvailableKeyIds() {
+        return new ArrayList<>(keys.keySet());
+    }
+
+    /**
+     * Adds the key to the provider and associates it with the given ID. Some implementations may not allow this operation.
+     *
+     * @param keyId the key identifier
+     * @param key   the key
+     * @return true if the key was successfully added
+     * @throws OperationNotSupportedException if this implementation doesn't support adding keys
+     * @throws KeyManagementException         if the key is invalid, the ID conflicts, etc.
+     */
+    @Override
+    public boolean addKey(String keyId, SecretKey key) throws OperationNotSupportedException, KeyManagementException {
+        throw new OperationNotSupportedException("This implementation does not allow adding keys");
+    }
+}

http://git-wip-us.apache.org/repos/asf/nifi/blob/7d242076/nifi-commons/nifi-data-provenance-utils/src/test/groovy/org/apache/nifi/provenance/AESProvenanceEventEncryptorTest.groovy
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-data-provenance-utils/src/test/groovy/org/apache/nifi/provenance/AESProvenanceEventEncryptorTest.groovy b/nifi-commons/nifi-data-provenance-utils/src/test/groovy/org/apache/nifi/provenance/AESProvenanceEventEncryptorTest.groovy
new file mode 100644
index 0000000..61b35d0
--- /dev/null
+++ b/nifi-commons/nifi-data-provenance-utils/src/test/groovy/org/apache/nifi/provenance/AESProvenanceEventEncryptorTest.groovy
@@ -0,0 +1,303 @@
+/*
+ * 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.nifi.provenance
+
+import org.apache.nifi.security.util.EncryptionMethod
+import org.apache.nifi.security.util.crypto.AESKeyedCipherProvider
+import org.bouncycastle.jce.provider.BouncyCastleProvider
+import org.bouncycastle.util.encoders.Hex
+import org.junit.After
+import org.junit.AfterClass
+import org.junit.Before
+import org.junit.BeforeClass
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.junit.runners.JUnit4
+import org.slf4j.Logger
+import org.slf4j.LoggerFactory
+
+import javax.crypto.Cipher
+import javax.crypto.SecretKey
+import javax.crypto.spec.IvParameterSpec
+import javax.crypto.spec.SecretKeySpec
+import java.nio.charset.StandardCharsets
+import java.security.KeyManagementException
+import java.security.SecureRandom
+import java.security.Security
+
+import static groovy.test.GroovyAssert.shouldFail
+
+@RunWith(JUnit4.class)
+class AESProvenanceEventEncryptorTest {
+    private static final Logger logger = LoggerFactory.getLogger(AESProvenanceEventEncryptorTest.class)
+
+    private static final String KEY_HEX_128 = "0123456789ABCDEFFEDCBA9876543210"
+    private static final String KEY_HEX_256 = KEY_HEX_128 * 2
+    private static final String KEY_HEX = isUnlimitedStrengthCryptoAvailable() ? KEY_HEX_256 : KEY_HEX_128
+
+    private static KeyProvider mockKeyProvider
+    private static AESKeyedCipherProvider mockCipherProvider
+
+    private static String ORIGINAL_LOG_LEVEL
+
+    private ProvenanceEventEncryptor encryptor
+
+    @BeforeClass
+    static void setUpOnce() throws Exception {
+        ORIGINAL_LOG_LEVEL = System.getProperty("org.slf4j.simpleLogger.log.org.apache.nifi.provenance")
+        System.setProperty("org.slf4j.simpleLogger.log.org.apache.nifi.provenance", "DEBUG")
+
+        Security.addProvider(new BouncyCastleProvider())
+
+        logger.metaClass.methodMissing = { String name, args ->
+            logger.info("[${name?.toUpperCase()}] ${(args as List).join(" ")}")
+        }
+
+        mockKeyProvider = [
+                getKey   : { String keyId ->
+                    logger.mock("Requesting key ID: ${keyId}")
+                    new SecretKeySpec(Hex.decode(KEY_HEX), "AES")
+                },
+                keyExists: { String keyId ->
+                    logger.mock("Checking existence of ${keyId}")
+                    true
+                }] as KeyProvider
+
+        mockCipherProvider = [
+                getCipher: { EncryptionMethod em, SecretKey key, byte[] ivBytes, boolean encryptMode ->
+                    logger.mock("Getting cipher for ${em} with IV ${Hex.toHexString(ivBytes)} encrypt ${encryptMode}")
+                    Cipher cipher = Cipher.getInstance("AES/GCM/NoPadding")
+                    cipher.init((encryptMode ? Cipher.ENCRYPT_MODE : Cipher.DECRYPT_MODE) as int, key, new IvParameterSpec(ivBytes))
+                    cipher
+                }
+        ] as AESKeyedCipherProvider
+    }
+
+    @Before
+    void setUp() throws Exception {
+
+    }
+
+    @After
+    void tearDown() throws Exception {
+
+    }
+
+    @AfterClass
+    static void tearDownOnce() throws Exception {
+        if (ORIGINAL_LOG_LEVEL) {
+            System.setProperty("org.slf4j.simpleLogger.log.org.apache.nifi.provenance", ORIGINAL_LOG_LEVEL)
+        }
+    }
+
+    private static boolean isUnlimitedStrengthCryptoAvailable() {
+        Cipher.getMaxAllowedKeyLength("AES") > 128
+    }
+
+    /**
+     * Given arbitrary bytes, encrypt them and persist with the encryption metadata, then recover
+     */
+    @Test
+    void testShouldEncryptAndDecryptArbitraryBytes() {
+        // Arrange
+        final byte[] SERIALIZED_BYTES = "This is a plaintext message.".getBytes(StandardCharsets.UTF_8)
+        logger.info("Serialized bytes (${SERIALIZED_BYTES.size()}): ${Hex.toHexString(SERIALIZED_BYTES)}")
+
+        encryptor = new AESProvenanceEventEncryptor()
+        encryptor.initialize(mockKeyProvider)
+        encryptor.setCipherProvider(mockCipherProvider)
+        logger.info("Created ${encryptor}")
+
+        String keyId = "K1"
+        String recordId = "R1"
+        logger.info("Using record ID ${recordId} and key ID ${keyId}")
+
+        // Act
+        byte[] metadataAndCipherBytes = encryptor.encrypt(SERIALIZED_BYTES, recordId, keyId)
+        logger.info("Encrypted data to: \n\t${Hex.toHexString(metadataAndCipherBytes)}")
+
+        byte[] recoveredBytes = encryptor.decrypt(metadataAndCipherBytes, recordId)
+        logger.info("Decrypted data to: \n\t${Hex.toHexString(recoveredBytes)}")
+
+        // Assert
+        assert recoveredBytes == SERIALIZED_BYTES
+        logger.info("Decoded (usually would be serialized schema record): ${new String(recoveredBytes, StandardCharsets.UTF_8)}")
+    }
+
+    @Test
+    void testShouldInitializeNullCipherProvider() {
+        // Arrange
+        encryptor = new AESProvenanceEventEncryptor()
+        encryptor.setCipherProvider(null)
+        assert !encryptor.aesKeyedCipherProvider
+        
+        // Act
+        encryptor.initialize(mockKeyProvider)
+        logger.info("Created ${encryptor}")
+
+        // Assert
+        assert encryptor.aesKeyedCipherProvider instanceof AESKeyedCipherProvider
+    }
+
+    @Test
+    void testShouldFailOnMissingKeyId() {
+        // Arrange
+        final byte[] SERIALIZED_BYTES = "This is a plaintext message.".getBytes(StandardCharsets.UTF_8)
+        logger.info("Serialized bytes (${SERIALIZED_BYTES.size()}): ${Hex.toHexString(SERIALIZED_BYTES)}")
+
+        KeyProvider emptyKeyProvider = [
+                getKey   : { String kid ->
+                    throw new KeyManagementException("No key found for ${kid}")
+                },
+                keyExists: { String kid -> false }
+        ] as KeyProvider
+
+        encryptor = new AESProvenanceEventEncryptor()
+        encryptor.initialize(emptyKeyProvider)
+        encryptor.setCipherProvider(mockCipherProvider)
+        logger.info("Created ${encryptor}")
+
+        String keyId = "K1"
+        String recordId = "R1"
+        logger.info("Using record ID ${recordId} and key ID ${keyId}")
+
+        // Act
+        def msg = shouldFail(EncryptionException) {
+            byte[] metadataAndCipherBytes = encryptor.encrypt(SERIALIZED_BYTES, recordId, keyId)
+            logger.info("Encrypted data to: \n\t${Hex.toHexString(metadataAndCipherBytes)}")
+        }
+        logger.expected(msg)
+
+        // Assert
+        assert msg.getMessage() == "The requested key ID is not available"
+    }
+
+    @Test
+    void testShouldUseDifferentIVsForSequentialEncryptions() {
+        // Arrange
+        final byte[] SERIALIZED_BYTES = "This is a plaintext message.".getBytes(StandardCharsets.UTF_8)
+        logger.info("Serialized bytes (${SERIALIZED_BYTES.size()}): ${Hex.toHexString(SERIALIZED_BYTES)}")
+
+        encryptor = new AESProvenanceEventEncryptor()
+        encryptor.initialize(mockKeyProvider)
+        logger.info("Created ${encryptor}")
+
+        String keyId = "K1"
+        String recordId1 = "R1"
+        logger.info("Using record ID ${recordId1} and key ID ${keyId}")
+
+        // Act
+        byte[] metadataAndCipherBytes1 = encryptor.encrypt(SERIALIZED_BYTES, recordId1, keyId)
+        logger.info("Encrypted data to: \n\t${Hex.toHexString(metadataAndCipherBytes1)}")
+        EncryptionMetadata metadata1 = encryptor.extractEncryptionMetadata(metadataAndCipherBytes1)
+        logger.info("Record ${recordId1} IV: ${Hex.toHexString(metadata1.ivBytes)}")
+
+        byte[] recoveredBytes1 = encryptor.decrypt(metadataAndCipherBytes1, recordId1)
+        logger.info("Decrypted data to: \n\t${Hex.toHexString(recoveredBytes1)}")
+
+        String recordId2 = "R2"
+        byte[] metadataAndCipherBytes2 = encryptor.encrypt(SERIALIZED_BYTES, recordId2, keyId)
+        logger.info("Encrypted data to: \n\t${Hex.toHexString(metadataAndCipherBytes2)}")
+        EncryptionMetadata metadata2 = encryptor.extractEncryptionMetadata(metadataAndCipherBytes2)
+        logger.info("Record ${recordId2} IV: ${Hex.toHexString(metadata2.ivBytes)}")
+
+        byte[] recoveredBytes2 = encryptor.decrypt(metadataAndCipherBytes2, recordId2)
+        logger.info("Decrypted data to: \n\t${Hex.toHexString(recoveredBytes2)}")
+
+        // Assert
+        assert metadata1.ivBytes != metadata2.ivBytes
+
+        assert recoveredBytes1 == SERIALIZED_BYTES
+        assert recoveredBytes2 == SERIALIZED_BYTES
+    }
+
+    @Test
+    void testShouldFailOnBadMetadata() {
+        // Arrange
+        final byte[] SERIALIZED_BYTES = "This is a plaintext message.".getBytes(StandardCharsets.UTF_8)
+        logger.info("Serialized bytes (${SERIALIZED_BYTES.size()}): ${Hex.toHexString(SERIALIZED_BYTES)}")
+
+        def strictMockKeyProvider = [
+                getKey   : { String keyId ->
+                    if (keyId != "K1") {
+                        throw new KeyManagementException("No such key")
+                    }
+                    new SecretKeySpec(Hex.decode(KEY_HEX), "AES")
+                },
+                keyExists: { String keyId ->
+                    keyId == "K1"
+                }] as KeyProvider
+
+        encryptor = new AESProvenanceEventEncryptor()
+        encryptor.initialize(strictMockKeyProvider)
+        encryptor.setCipherProvider(mockCipherProvider)
+        logger.info("Created ${encryptor}")
+
+        String keyId = "K1"
+        String recordId = "R1"
+        logger.info("Using record ID ${recordId} and key ID ${keyId}")
+
+        final String ALGORITHM = "AES/GCM/NoPadding"
+        final byte[] ivBytes = new byte[16]
+        new SecureRandom().nextBytes(ivBytes)
+        final String VERSION = "v1"
+
+        // Perform the encryption independently of the encryptor
+        SecretKey key = mockKeyProvider.getKey(keyId)
+        Cipher cipher = new AESKeyedCipherProvider().getCipher(EncryptionMethod.AES_GCM, key, ivBytes, true)
+        byte[] cipherBytes = cipher.doFinal(SERIALIZED_BYTES)
+
+        int cipherBytesLength = cipherBytes.size()
+
+        // Construct accurate metadata
+        EncryptionMetadata goodMetadata = new EncryptionMetadata(keyId, ALGORITHM, ivBytes, VERSION, cipherBytesLength)
+        logger.info("Created good encryption metadata: ${goodMetadata}")
+
+        // Construct bad metadata instances
+        EncryptionMetadata badKeyId = new EncryptionMetadata(keyId.reverse(), ALGORITHM, ivBytes, VERSION, cipherBytesLength)
+        EncryptionMetadata badAlgorithm = new EncryptionMetadata(keyId, "ASE/GDM/SomePadding", ivBytes, VERSION, cipherBytesLength)
+        EncryptionMetadata badIV = new EncryptionMetadata(keyId, ALGORITHM, new byte[16], VERSION, cipherBytesLength)
+        EncryptionMetadata badVersion = new EncryptionMetadata(keyId, ALGORITHM, ivBytes, VERSION.reverse(), cipherBytesLength)
+        EncryptionMetadata badCBLength = new EncryptionMetadata(keyId, ALGORITHM, ivBytes, VERSION, cipherBytesLength - 5)
+
+        List badMetadata = [badKeyId, badAlgorithm, badIV, badVersion, badCBLength]
+
+        // Form the proper cipherBytes
+        byte[] completeGoodBytes = CryptoUtils.concatByteArrays([0x01] as byte[], encryptor.serializeEncryptionMetadata(goodMetadata), cipherBytes)
+
+        byte[] recoveredGoodBytes = encryptor.decrypt(completeGoodBytes, recordId)
+        logger.info("Recovered good bytes: ${Hex.toHexString(recoveredGoodBytes)}")
+
+        final List EXPECTED_MESSAGES = ["The requested key ID (\\w+)? is not available",
+                                        "Encountered an exception decrypting provenance record",
+                                        "The event was encrypted with version ${VERSION.reverse()} which is not in the list of supported versions v1"]
+
+        // Act
+        badMetadata.eachWithIndex { EncryptionMetadata metadata, int i ->
+            byte[] completeBytes = CryptoUtils.concatByteArrays([0x01] as byte[], encryptor.serializeEncryptionMetadata(metadata), cipherBytes)
+
+            def msg = shouldFail(EncryptionException) {
+                byte[] recoveredBytes = encryptor.decrypt(completeBytes, "R${i + 2}")
+                logger.info("Recovered bad bytes: ${Hex.toHexString(recoveredBytes)}")
+            }
+            logger.expected(msg)
+
+            // Assert
+            assert EXPECTED_MESSAGES.any { msg.getMessage() =~ it }
+        }
+    }
+}


Mime
View raw message