NIFI-6325 Added AWS KMS Sensitive Properties Provider

This closes #5202

Signed-off-by: David Handermann <exceptionfactory@apache.org>

nifi-assembly/NOTICE

@@ -1013,6 +1013,13 @@ The following binary components are provided under the Apache Software License v
       Since product implements StAX API, it has dependencies to StAX API
       classes.
 
+  (ASLv2) AWS SDK for Java 2.0
+      The following NOTICE information applies:
+        Copyright 2010-2021 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+
+        This product includes software developed by
+        Amazon Technologies, Inc (https://www.amazon.com/).
+
   (ASLv2) Amazon Web Services SDK
     The following NOTICE information applies:
       Copyright 2010-2014 Amazon.com, Inc. or its affiliates. All Rights Reserved.

nifi-commons/nifi-property-utils/src/main/java/org/apache/nifi/properties/BootstrapProperties.java

@@ -32,7 +32,8 @@ public class BootstrapProperties extends StandardReadableProperties {
 
     public enum BootstrapPropertyKey {
         SENSITIVE_KEY("bootstrap.sensitive.key"),
-        HASHICORP_VAULT_SENSITIVE_PROPERTY_PROVIDER_CONF("bootstrap.protection.hashicorp.vault.conf");
+        HASHICORP_VAULT_SENSITIVE_PROPERTY_PROVIDER_CONF("bootstrap.protection.hashicorp.vault.conf"),
+        AWS_KMS_SENSITIVE_PROPERTY_PROVIDER_CONF("bootstrap.protection.aws.kms.conf");
 
         private final String key;
 

nifi-commons/nifi-sensitive-property-provider/pom.xml

@@ -21,7 +21,9 @@
         <version>1.15.0-SNAPSHOT</version>
     </parent>
     <artifactId>nifi-sensitive-property-provider</artifactId>
-
+    <properties>
+        <aws.sdk.version>2.17.1</aws.sdk.version>
+    </properties>
     <dependencies>
         <dependency>
             <groupId>org.apache.nifi</groupId>
@@ -42,6 +44,26 @@
             <artifactId>nifi-security-utils</artifactId>
             <version>1.15.0-SNAPSHOT</version>
         </dependency>
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>kms</artifactId>
+            <version>${aws.sdk.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>software.amazon.awssdk</groupId>
+                    <artifactId>netty-nio-client</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>software.amazon.awssdk</groupId>
+                    <artifactId>apache-client</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>url-connection-client</artifactId>
+            <version>${aws.sdk.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.nifi</groupId>
             <artifactId>nifi-vault-utils</artifactId>

nifi-commons/nifi-sensitive-property-provider/src/main/java/org/apache/nifi/properties/AESSensitivePropertyProvider.java

@@ -257,4 +257,10 @@ public class AESSensitivePropertyProvider extends AbstractSensitivePropertyProvi
     public static String getDelimiter() {
         return DELIMITER;
     }
+
+    /**
+     * No cleanup necessary
+     */
+    @Override
+    public void cleanUp() { }
 }

nifi-commons/nifi-sensitive-property-provider/src/main/java/org/apache/nifi/properties/AWSSensitivePropertyProvider.java

@@ -0,0 +1,336 @@
+/*
+ * 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.properties;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.properties.BootstrapProperties.BootstrapPropertyKey;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.core.exception.SdkClientException;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.kms.KmsClient;
+import software.amazon.awssdk.services.kms.model.DecryptRequest;
+import software.amazon.awssdk.services.kms.model.DecryptResponse;
+import software.amazon.awssdk.services.kms.model.DescribeKeyRequest;
+import software.amazon.awssdk.services.kms.model.DescribeKeyResponse;
+import software.amazon.awssdk.services.kms.model.EncryptRequest;
+import software.amazon.awssdk.services.kms.model.EncryptResponse;
+import software.amazon.awssdk.services.kms.model.KeyMetadata;
+import software.amazon.awssdk.services.kms.model.KmsException;
+
+import java.util.Base64;
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Paths;
+import java.util.Objects;
+
+public class AWSSensitivePropertyProvider extends AbstractSensitivePropertyProvider {
+    private static final Logger logger = LoggerFactory.getLogger(AWSSensitivePropertyProvider.class);
+
+    private static final String AWS_PREFIX = "aws";
+    private static final String ACCESS_KEY_PROPS_NAME = "aws.access.key.id";
+    private static final String SECRET_KEY_PROPS_NAME = "aws.secret.access.key";
+    private static final String REGION_KEY_PROPS_NAME = "aws.region";
+    private static final String KMS_KEY_PROPS_NAME = "aws.kms.key.id";
+
+    private static final Charset PROPERTY_CHARSET = StandardCharsets.UTF_8;
+
+    private final BootstrapProperties awsBootstrapProperties;
+    private KmsClient client;
+    private String keyId;
+
+
+    AWSSensitivePropertyProvider(final BootstrapProperties bootstrapProperties) throws SensitivePropertyProtectionException {
+        super(bootstrapProperties);
+        Objects.requireNonNull(bootstrapProperties, "The file bootstrap.conf provided to AWS SPP is null");
+        awsBootstrapProperties = getAWSBootstrapProperties(bootstrapProperties);
+        loadRequiredAWSProperties(awsBootstrapProperties);
+    }
+
+    /**
+     * Initializes the KMS Client to be used for encrypt, decrypt and other interactions with AWS KMS.
+     * First attempts to use credentials/configuration in bootstrap-aws.conf.
+     * If credentials/configuration in bootstrap-aws.conf is not fully configured,
+     * attempt to initialize credentials using default AWS credentials/configuration chain.
+     * Note: This does not verify if credentials are valid.
+     */
+    private void initializeClient() {
+        if (awsBootstrapProperties == null) {
+            logger.warn("AWS Bootstrap Properties are required for KMS Client initialization");
+            return;
+        }
+        final String accessKey = awsBootstrapProperties.getProperty(ACCESS_KEY_PROPS_NAME);
+        final String secretKey = awsBootstrapProperties.getProperty(SECRET_KEY_PROPS_NAME);
+        final String region = awsBootstrapProperties.getProperty(REGION_KEY_PROPS_NAME);
+
+        if (StringUtils.isNoneBlank(accessKey, secretKey, region)) {
+            logger.debug("Using AWS credentials from bootstrap properties");
+            try {
+                final AwsBasicCredentials credentials = AwsBasicCredentials.create(accessKey, secretKey);
+                client = KmsClient.builder()
+                        .region(Region.of(region))
+                        .credentialsProvider(StaticCredentialsProvider.create(credentials))
+                        .build();
+            } catch (final RuntimeException e) {
+                final String msg = "Valid configuration/credentials are required to initialize KMS client";
+                throw new SensitivePropertyProtectionException(msg, e);
+            }
+        } else {
+            logger.debug("Using AWS credentials from default credentials provider");
+            try {
+                final DefaultCredentialsProvider credentialsProvider = DefaultCredentialsProvider.builder()
+                        .build();
+                credentialsProvider.resolveCredentials();
+                client = KmsClient.builder()
+                        .credentialsProvider(credentialsProvider)
+                        .build();
+            } catch (final SdkClientException e) {
+                final String msg = "Valid configuration/credentials are required to initialize KMS client";
+                throw new SensitivePropertyProtectionException(msg, e);
+            }
+        }
+    }
+
+    /**
+     * Validates the key ARN, credentials and configuration provided by the user.
+     * Note: This function performs checks on the key and indirectly also validates the credentials and
+     * configurations provided during the initialization of the client.
+     */
+    private void validate() throws KmsException, SensitivePropertyProtectionException {
+        if (client == null) {
+            final String msg = "The AWS KMS Client failed to open, cannot validate key";
+            throw new SensitivePropertyProtectionException(msg);
+        }
+        if (StringUtils.isBlank(keyId)) {
+            final String msg = "The AWS KMS key provided is blank";
+            throw new SensitivePropertyProtectionException(msg);
+        }
+
+        // asking for a Key Description is the best way to check whether a key is valid
+        // because AWS KMS accepts various formats for its keys.
+        final DescribeKeyRequest request = DescribeKeyRequest.builder()
+                .keyId(keyId)
+                .build();
+
+        // using the KmsClient in a DescribeKey request indirectly also verifies if the credentials provided
+        // during the initialization of the key are valid
+        final DescribeKeyResponse response = client.describeKey(request);
+        final KeyMetadata metadata = response.keyMetadata();
+
+        if (!metadata.enabled()) {
+            final String msg = String.format("AWS KMS key [%s] is not enabled", keyId);
+            throw new SensitivePropertyProtectionException(msg);
+        }
+    }
+
+    /**
+     * Checks if we have a key ID from AWS KMS and loads it into {@link #keyId}. Will load null if key is not present.
+     * Note: This function does not verify if the key is correctly formatted/valid.
+     * @param props the properties representing bootstrap-aws.conf
+     */
+    private void loadRequiredAWSProperties(final BootstrapProperties props) {
+        if (props != null) {
+            keyId = props.getProperty(KMS_KEY_PROPS_NAME);
+        }
+    }
+
+
+    /**
+     * Checks bootstrap.conf to check if BootstrapPropertyKey.AWS_KMS_SENSITIVE_PROPERTY_PROVIDER_CONF property is
+     * configured to the bootstrap-aws.conf file. Also will try to load bootstrap-aws.conf to {@link #awsBootstrapProperties}.
+     * @param bootstrapProperties BootstrapProperties object corresponding to bootstrap.conf.
+     * @return BootstrapProperties object corresponding to bootstrap-aws.conf, null otherwise.
+     */
+    private BootstrapProperties getAWSBootstrapProperties(final BootstrapProperties bootstrapProperties) {
+        final BootstrapProperties cloudBootstrapProperties;
+
+        // Load the bootstrap-aws.conf file based on path specified in
+        // "nifi.bootstrap.protection.aws.kms.conf" property of bootstrap.conf
+        final String filePath = bootstrapProperties.getProperty(BootstrapPropertyKey.AWS_KMS_SENSITIVE_PROPERTY_PROVIDER_CONF).orElse(null);
+        if (StringUtils.isBlank(filePath)) {
+            logger.warn("AWS KMS properties file path not configured in bootstrap properties");
+            return null;
+        }
+
+        try {
+            cloudBootstrapProperties = AbstractBootstrapPropertiesLoader.loadBootstrapProperties(
+                    Paths.get(filePath), AWS_PREFIX);
+        } catch (final IOException e) {
+            throw new SensitivePropertyProtectionException("Could not load " + filePath, e);
+        }
+
+        return cloudBootstrapProperties;
+    }
+
+    /**
+     * Checks bootstrap-aws.conf for the required configurations for AWS KMS encrypt/decrypt operations.
+     * Note: This does not check for credentials/region configurations.
+     *       Credentials/configuration will be checked during the first protect/unprotect call during runtime.
+     * @return true if bootstrap-aws.conf contains the required properties for AWS SPP, false otherwise.
+     */
+    private boolean hasRequiredAWSProperties() {
+        return awsBootstrapProperties != null && StringUtils.isNotBlank(keyId);
+    }
+
+    @Override
+    public boolean isSupported() {
+        return hasRequiredAWSProperties();
+    }
+
+    @Override
+    protected PropertyProtectionScheme getProtectionScheme() {
+        return PropertyProtectionScheme.AWS_KMS;
+    }
+
+    /**
+     * Returns the name of the underlying implementation.
+     *
+     * @return the name of this sensitive property provider.
+     */
+    @Override
+    public String getName() {
+        return PropertyProtectionScheme.AWS_KMS.getName();
+    }
+
+    /**
+     * Returns the key used to identify the provider implementation in {@code nifi.properties}.
+     *
+     * @return the key to persist in the sibling property.
+     */
+    @Override
+    public String getIdentifierKey() {
+        return PropertyProtectionScheme.AWS_KMS.getIdentifier();
+    }
+
+
+    /**
+     * Returns the ciphertext blob of this value encrypted using an AWS KMS CMK.
+     *
+     * @return the ciphertext blob to persist in the {@code nifi.properties} file.
+     */
+    private byte[] encrypt(final byte[] input) {
+        final SdkBytes plainBytes = SdkBytes.fromByteArray(input);
+
+        final EncryptRequest encryptRequest = EncryptRequest.builder()
+                .keyId(keyId)
+                .plaintext(plainBytes)
+                .build();
+
+        final EncryptResponse response = client.encrypt(encryptRequest);
+        final SdkBytes encryptedData = response.ciphertextBlob();
+
+        return encryptedData.asByteArray();
+    }
+
+    /**
+     * Returns the value corresponding to a ciphertext blob decrypted using an AWS KMS CMK.
+     *
+     * @return the "unprotected" byte[] of this value, which could be used by the application.
+     */
+    private byte[] decrypt(final byte[] input) {
+        final SdkBytes cipherBytes = SdkBytes.fromByteArray(input);
+
+        final DecryptRequest decryptRequest = DecryptRequest.builder()
+                .ciphertextBlob(cipherBytes)
+                .keyId(keyId)
+                .build();
+
+        final DecryptResponse response = client.decrypt(decryptRequest);
+        final SdkBytes decryptedData = response.plaintext();
+
+        return decryptedData.asByteArray();
+    }
+
+    /**
+     * Checks if the client is open and if not, initializes the client and validates the key required for AWS KMS.
+     */
+    private void checkAndInitializeClient() throws SensitivePropertyProtectionException {
+        if (client == null) {
+            try {
+                initializeClient();
+                validate();
+            } catch (final SdkClientException | KmsException | SensitivePropertyProtectionException e) {
+                throw new SensitivePropertyProtectionException("Error initializing the AWS KMS Client", e);
+            }
+        }
+    }
+
+    /**
+     * Returns the "protected" form of this value. This is a form which can safely be persisted in the {@code nifi.properties} file without compromising the value.
+     * An encryption-based provider would return a cipher text, while a remote-lookup provider could return a unique ID to retrieve the secured value.
+     *
+     * @param unprotectedValue the sensitive value.
+     * @return the value to persist in the {@code nifi.properties} file.
+     */
+    @Override
+    public String protect(final String unprotectedValue) throws SensitivePropertyProtectionException {
+        if (StringUtils.isBlank(unprotectedValue)) {
+            throw new IllegalArgumentException("Cannot encrypt a blank value");
+        }
+
+        checkAndInitializeClient();
+
+        try {
+            final byte[] plainBytes = unprotectedValue.getBytes(PROPERTY_CHARSET);
+            final byte[] cipherBytes = encrypt(plainBytes);
+            return Base64.getEncoder().encodeToString(cipherBytes);
+        } catch (final SdkClientException | KmsException e) {
+            throw new SensitivePropertyProtectionException("Encrypt failed", e);
+        }
+    }
+
+    /**
+     * Returns the "unprotected" form of this value. This is the raw sensitive value which is used by the application logic.
+     * An encryption-based provider would decrypt a cipher text and return the plaintext, while a remote-lookup provider could retrieve the secured value.
+     *
+     * @param protectedValue the protected value read from the {@code nifi.properties} file.
+     * @return the raw value to be used by the application.
+     */
+    @Override
+    public String unprotect(final String protectedValue) throws SensitivePropertyProtectionException {
+        if (StringUtils.isBlank(protectedValue)) {
+            throw new IllegalArgumentException("Cannot decrypt a blank value");
+        }
+
+        checkAndInitializeClient();
+
+        try {
+            final byte[] cipherBytes = Base64.getDecoder().decode(protectedValue);
+            final byte[] plainBytes = decrypt(cipherBytes);
+            return new String(plainBytes, PROPERTY_CHARSET);
+        } catch (final SdkClientException | KmsException e) {
+            throw new SensitivePropertyProtectionException("Decrypt failed", e);
+        }
+    }
+
+    /**
+     * Closes AWS KMS client that may have been opened.
+     */
+    @Override
+    public void cleanUp() {
+        if (client != null) {
+            client.close();
+            client = null;
+        }
+    }
+}

nifi-commons/nifi-sensitive-property-provider/src/main/java/org/apache/nifi/properties/AbstractHashiCorpVaultSensitivePropertyProvider.java

@@ -129,4 +129,9 @@ public abstract class AbstractHashiCorpVaultSensitivePropertyProvider extends Ab
         return getProtectionScheme().getIdentifier(path);
     }
 
+    /**
+     * No cleanup necessary
+     */
+    @Override
+    public void cleanUp() { }
 }

nifi-commons/nifi-sensitive-property-provider/src/main/java/org/apache/nifi/properties/PropertyProtectionScheme.java

@@ -25,6 +25,7 @@ import java.util.Objects;
  */
 public enum PropertyProtectionScheme {
     AES_GCM("aes/gcm/(128|192|256)", "aes/gcm/%s", "AES Sensitive Property Provider", true),
+    AWS_KMS("aws/kms", "aws/kms", "AWS KMS Sensitive Property Provider", false),
     HASHICORP_VAULT_TRANSIT("hashicorp/vault/transit/[a-zA-Z0-9_-]+", "hashicorp/vault/transit/%s", "HashiCorp Vault Transit Engine Sensitive Property Provider", false);
 
     PropertyProtectionScheme(final String identifierPattern, final String identifierFormat, final String name, final boolean requiresSecretKey) {

nifi-commons/nifi-sensitive-property-provider/src/main/java/org/apache/nifi/properties/SensitivePropertyProvider.java

@@ -56,4 +56,9 @@ public interface SensitivePropertyProvider {
      * @return the raw value to be used by the application
      */
     String unprotect(String protectedValue) throws SensitivePropertyProtectionException;
+
+    /**
+     * Cleans up resources that may have been allocated/used by an SPP implementation
+     */
+    void cleanUp();
 }

nifi-commons/nifi-sensitive-property-provider/src/main/java/org/apache/nifi/properties/StandardSensitivePropertyProviderFactory.java

@@ -105,6 +105,8 @@ public class StandardSensitivePropertyProviderFactory implements SensitiveProper
         switch (protectionScheme) {
             case AES_GCM:
                 return providerMap.computeIfAbsent(protectionScheme, s -> new AESSensitivePropertyProvider(keyHex));
+            case AWS_KMS:
+                return providerMap.computeIfAbsent(protectionScheme, s -> new AWSSensitivePropertyProvider(getBootstrapProperties()));
             case HASHICORP_VAULT_TRANSIT:
                 return providerMap.computeIfAbsent(protectionScheme, s -> new HashiCorpVaultTransitSensitivePropertyProvider(getBootstrapProperties()));
             default:

nifi-commons/nifi-sensitive-property-provider/src/test/java/org/apache/nifi/properties/AWSSensitivePropertyProviderIT.java

@@ -0,0 +1,133 @@
+/*
+ * 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.properties;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.mockito.internal.util.io.IOUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Properties;
+
+/**
+ * To run this test, make sure to first configure sensitive credential information as in the following link
+ * https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-files.html
+ *
+ * If you don't have a key then run:
+ * aws kms create-key
+ *
+ * Take note of the key id or arn.
+ *
+ * Then, set the system property -Daws.kms.key.id to the either key id value or arn value
+ *
+ * The following settings are optional. If you have a default AWS configuration and credentials in ~/.aws then
+ * it will take that. Otherwise you can set all of the following:
+ * set the system property -Daws.access.key.id to the access key id
+ * set the system property -Daws.secret.access.key to the secret access key
+ * set the system property -Daws.region to the region
+ *
+ * After you are satisfied with the test, and you don't need the key, you may schedule key deletion with:
+ * aws kms schedule-key-deletion --key-id "key id" --pending-window-in-days "number of days"
+ *
+ */
+
+public class AWSSensitivePropertyProviderIT {
+    private static final String SAMPLE_PLAINTEXT = "AWSSensitivePropertyProviderIT SAMPLE-PLAINTEXT";
+    private static final String ACCESS_KEY_PROPS_NAME = "aws.access.key.id";
+    private static final String SECRET_KEY_PROPS_NAME = "aws.secret.access.key";
+    private static final String REGION_KEY_PROPS_NAME = "aws.region";
+    private static final String KMS_KEY_PROPS_NAME = "aws.kms.key.id";
+
+    private static final String BOOTSTRAP_AWS_FILE_PROPS_NAME = "nifi.bootstrap.protection.aws.kms.conf";
+
+    private static final String EMPTY_PROPERTY = "";
+
+    private static AWSSensitivePropertyProvider spp;
+
+    private static BootstrapProperties props;
+
+    private static Path mockBootstrapConf, mockAWSBootstrapConf;
+
+    private static final Logger logger = LoggerFactory.getLogger(AWSSensitivePropertyProviderIT.class);
+
+    private static void initializeBootstrapProperties() throws IOException{
+        mockBootstrapConf = Files.createTempFile("bootstrap", ".conf").toAbsolutePath();
+        mockAWSBootstrapConf = Files.createTempFile("bootstrap-aws", ".conf").toAbsolutePath();
+        IOUtil.writeText(BOOTSTRAP_AWS_FILE_PROPS_NAME + "=" + mockAWSBootstrapConf.toAbsolutePath(), mockBootstrapConf.toFile());
+
+        final Properties bootstrapProperties = new Properties();
+        try (final InputStream inputStream = Files.newInputStream(mockBootstrapConf)) {
+            bootstrapProperties.load(inputStream);
+            props = new BootstrapProperties("nifi", bootstrapProperties, mockBootstrapConf);
+        }
+
+        String accessKey = System.getProperty(ACCESS_KEY_PROPS_NAME, EMPTY_PROPERTY);
+        String secretKey = System.getProperty(SECRET_KEY_PROPS_NAME, EMPTY_PROPERTY);
+        String region = System.getProperty(REGION_KEY_PROPS_NAME, EMPTY_PROPERTY);
+        String keyId = System.getProperty(KMS_KEY_PROPS_NAME, EMPTY_PROPERTY);
+
+        StringBuilder bootstrapConfText = new StringBuilder();
+        bootstrapConfText.append(ACCESS_KEY_PROPS_NAME + "=" + accessKey);
+        bootstrapConfText.append("\n" + SECRET_KEY_PROPS_NAME + "=" + secretKey);
+        bootstrapConfText.append("\n" + REGION_KEY_PROPS_NAME + "=" + region);
+        bootstrapConfText.append("\n" + KMS_KEY_PROPS_NAME + "=" + keyId);
+
+        IOUtil.writeText(bootstrapConfText.toString(), mockAWSBootstrapConf.toFile());
+    }
+
+    @BeforeClass
+    public static void initOnce() throws IOException {
+        initializeBootstrapProperties();
+        Assert.assertNotNull(props);
+        spp = new AWSSensitivePropertyProvider(props);
+        Assert.assertNotNull(spp);
+    }
+
+    @AfterClass
+    public static void tearDownOnce() throws IOException {
+        Files.deleteIfExists(mockBootstrapConf);
+        Files.deleteIfExists(mockAWSBootstrapConf);
+
+        spp.cleanUp();
+    }
+
+    @Test
+    public void testEncryptDecrypt() {
+        logger.info("Running testEncryptDecrypt of AWS SPP integration test");
+        runEncryptDecryptTest();
+        logger.info("testEncryptDecrypt of AWS SPP integration test completed");
+    }
+
+    private static void runEncryptDecryptTest() {
+        logger.info("Plaintext: " + SAMPLE_PLAINTEXT);
+        String protectedValue = spp.protect(SAMPLE_PLAINTEXT);
+        logger.info("Protected Value: " + protectedValue);
+        String unprotectedValue = spp.unprotect(protectedValue);
+        logger.info("Unprotected Value: " + unprotectedValue);
+
+        Assert.assertEquals(SAMPLE_PLAINTEXT, unprotectedValue);
+        Assert.assertNotEquals(SAMPLE_PLAINTEXT, protectedValue);
+        Assert.assertNotEquals(protectedValue, unprotectedValue);
+    }
+}

nifi-docs/src/main/asciidoc/toolkit-guide.adoc

@@ -435,7 +435,7 @@ The following are available options when targeting NiFi:
  * `-f`,`--flowXml <file>`                       The _flow.xml.gz_ file currently protected with old password (will be overwritten unless `-g` is specified)
  * `-g`,`--outputFlowXml <file>`                 The destination _flow.xml.gz_ file containing protected config values (will not modify input _flow.xml.gz_)
  * `-b`,`--bootstrapConf <file>`                 The bootstrap.conf file to persist root key and to optionally provide any configuration for the protection scheme.
- * `-S`,`--protectionScheme <protectionScheme>`  Selects the protection scheme for encrypted properties.  Valid values are: [AES_GCM, HASHICORP_VAULT_TRANSIT] (default is AES_GCM)
+ * `-S`,`--protectionScheme <protectionScheme>`  Selects the protection scheme for encrypted properties.  Valid values are: [AES_GCM, HASHICORP_VAULT_TRANSIT, AWS_KMS] (default is AES_GCM)
  * `-k`,`--key <keyhex>`                         The raw hexadecimal key to use to encrypt the sensitive properties
  * `-e`,`--oldKey <keyhex>`                      The old raw hexadecimal key to use during key migration
  * `-H`,`--oldProtectionScheme <protectionScheme>` The old protection scheme to use during encryption migration (see --protectionScheme for possible values).  Default is AES_GCM
@@ -456,7 +456,7 @@ The following are available options when targeting NiFi Registry using the `--ni
  * `-v`,`--verbose`                              Sets verbose mode (default false)
  * `-p`,`--password <password>`                  Protect the files using a password-derived key. If an argument is not provided to this flag, interactive mode will be triggered to prompt the user to enter the password.
  * `-k`,`--key <keyhex>`                         Protect the files using a raw hexadecimal key. If an argument is not provided to this flag, interactive mode will be triggered to prompt the user to enter the key.
- * `-S`,`--protectionScheme <protectionScheme>`  Selects the protection scheme for encrypted properties.  Valid values are: [AES_GCM, HASHICORP_VAULT_TRANSIT]  (default is AES_GCM)
+ * `-S`,`--protectionScheme <protectionScheme>`  Selects the protection scheme for encrypted properties.  Valid values are: [AES_GCM, HASHICORP_VAULT_TRANSIT, AWS_KMS]  (default is AES_GCM)
  * `--oldPassword <password>`                    If the input files are already protected using a password-derived key, this specifies the old password so that the files can be unprotected before re-protecting.
  * `--oldKey <keyhex>`                           If the input files are already protected using a key, this specifies the raw hexadecimal key so that the files can be unprotected before re-protecting.
  * `-H`,`--oldProtectionScheme <protectionScheme>`The old protection scheme to use during encryption migration (see --protectionScheme for possible values).  Default is AES_GCM.
@@ -504,6 +504,26 @@ This protection scheme uses HashiCorp Vault's Transit Secrets Engine (https://ww
 |`vault.ssl.trust-store-password`|Truststore password.  Required if the Vault server is TLS-enabled|_none_
 |===
 
+==== AWS_KMS
+This protection scheme uses AWS Key Management Service (https://aws.amazon.com/kms/) for encryption and decryption. AWS KMS configuration properties can be stored in the `bootstrap-aws.conf` file, as referenced in the `bootstrap.conf` of NiFi or NiFi Registry. If the configuration properties are not specified in `bootstrap-aws.conf`, then the provider will attempt to use the AWS default credentials provider, which checks standard environment variables and system properties.
+
+===== Required properties
+[options="header,footer"]
+|===
+|Property Name|Description|Default
+|`aws.kms.key.id`|The identifier or ARN that the AWS KMS client uses for encryption and decryption.|_none_
+|===
+
+===== Optional properties
+====== All of the following must be configured, or will be ignored entirely.
+[options="header,footer"]
+|===
+|Property Name|Description|Default
+|`aws.region`|The AWS region used to configure the AWS KMS Client.|_none_
+|`aws.access.key.id`|The access key ID credential used to access AWS KMS.|_none_
+|`aws.secret.access.key`|The secret access key used to access AWS KMS.|_none_
+|===
+
 === Examples
 
 ==== NiFi
@@ -699,6 +719,40 @@ for each phase (old vs. new), and any combination is sufficient:
 In order to change the protection scheme (e.g., migrating from AES encryption to Vault encryption), specify the `--protectionScheme`
 and `--oldProtectionScheme` in the migration command.
 
+The following is an example of the commands for protection scheme migration from AES_GCM to AWS_KMS then back. Execute these commands at the `nifi` directory with the `nifi-toolkit` directory as a sibling directory. In addition, make sure to update `bootstrap-aws.conf` with your AWS KMS Key ARN/ID and have your credentials and region configured.
+
+
+This command encrypts nifi.properties with the AES_GCM protection scheme
+----
+./../nifi-toolkit-*-SNAPSHOT/bin/encrypt-config.sh \
+-b conf/bootstrap.conf \
+-n conf/nifi.properties \
+-k 0123456789ABCDEFFEDCBA98765432100123456789ABCDEFFEDCBA9876543210 \
+-v
+----
+This command migrates nifi.properties from using AES_GCM to using AWS_KMS protection scheme
+----
+./../nifi-toolkit-*-SNAPSHOT/bin/encrypt-config.sh \
+-b conf/bootstrap.conf \
+-n conf/nifi.properties \
+-S AWS_KMS \
+-H AES_GCM \
+-e 0123456789ABCDEFFEDCBA98765432100123456789ABCDEFFEDCBA9876543210 \
+-m \
+-v
+----
+This command migrates nifi.properties back from AWS_KMS to AES_GCM protection scheme
+----
+./../nifi-toolkit-*-SNAPSHOT/bin/encrypt-config.sh \
+-b conf/bootstrap.conf \
+-n conf/nifi.properties \
+-S AES_GCM \
+-k 0123456789ABCDEFFEDCBA98765432100123456789ABCDEFFEDCBA9876543210 \
+-H AWS_KMS \
+-m \
+-v
+----
+
 == File Manager
 The File Manager utility (invoked as `./bin/file-manager.sh` or `bin\file-manager.bat`) allows system administrators to take a backup of an existing NiFi installation, install a new version of NiFi in a designated location (while migrating any previous configuration settings) or restore an installation from a previous backup. File Manager supports NiFi version 1.0.0 and higher.
 

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-properties-loader/src/main/java/org/apache/nifi/properties/NiFiPropertiesLoader.java

@@ -173,8 +173,13 @@ public class NiFiPropertiesLoader {
                     .getSupportedSensitivePropertyProviders()
                     .forEach(protectedNiFiProperties::addSensitivePropertyProvider);
         }
-
-        return protectedNiFiProperties.getUnprotectedProperties();
+        NiFiProperties props = protectedNiFiProperties.getUnprotectedProperties();
+        if (protectedNiFiProperties.hasProtectedKeys()) {
+            getSensitivePropertyProviderFactory()
+                    .getSupportedSensitivePropertyProviders()
+                    .forEach(SensitivePropertyProvider::cleanUp);
+        }
+        return props;
     }
 
     /**

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/bootstrap-aws.conf

@@ -0,0 +1,27 @@
+#
+# 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.
+#
+
+# AWS KMS Key ID is required to be configured for AWS KMS Sensitive Property Provider
+aws.kms.key.id=
+
+# NiFi uses the following properties when authentication to AWS when all values are provided.
+# NiFi uses the default AWS credentials provider chain when one or more or the following properties are blank
+# AWS SDK documentation describes the default credential retrieval order:
+# https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/credentials.html#credentials-chain
+aws.access.key.id=
+aws.secret.access.key=
+aws.region=

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/bootstrap.conf

@@ -63,6 +63,9 @@ nifi.bootstrap.sensitive.key=
 # HashiCorp Vault Sensitive Property Providers
 nifi.bootstrap.protection.hashicorp.vault.conf=./conf/bootstrap-hashicorp-vault.conf
 
+# AWS KMS Sensitive Property Providers
+nifi.bootstrap.protection.aws.kms.conf=./conf/bootstrap-aws.conf
+
 # Sets the provider of SecureRandom to /dev/urandom to prevent blocking on VMs
 java.arg.15=-Djava.security.egd=file:/dev/urandom
 

nifi-nar-bundles/nifi-grpc-bundle/nifi-grpc-processors/pom.xml

@@ -75,6 +75,13 @@ language governing permissions and limitations under the License. -->
             <artifactId>nifi-mock</artifactId>
             <version>1.15.0-SNAPSHOT</version>
             <scope>test</scope>
+            <!-- Exclude transitive dependency of software.amazon.awssdk:kms included in nifi-sensitive-property-provider -->
+            <exclusions>
+                <exclusion>
+                    <artifactId>netty-transport-native-epoll</artifactId>
+                    <groupId>io.netty</groupId>
+                </exclusion>
+            </exclusions>
         </dependency>
         <dependency>
             <groupId>org.apache.nifi</groupId>

nifi-registry/nifi-registry-core/nifi-registry-resources/src/main/resources/conf/bootstrap-aws.conf

@@ -0,0 +1,27 @@
+#
+# 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.
+#
+
+# AWS KMS Key ID is required to be configured for AWS KMS Sensitive Property Provider
+aws.kms.key.id=
+
+# NiFi uses the following properties when authentication to AWS when all values are provided.
+# NiFi uses the default AWS credentials provider chain when one or more or the following properties are blank
+# AWS SDK documentation describes the default credential retrieval order:
+# https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/credentials.html#credentials-chain
+aws.access.key.id=
+aws.secret.access.key=
+aws.region=

nifi-registry/nifi-registry-core/nifi-registry-resources/src/main/resources/conf/bootstrap.conf

@@ -57,3 +57,6 @@ nifi.registry.bootstrap.sensitive.key=
 
 # HashiCorp Vault Sensitive Property Providers
 nifi.registry.bootstrap.protection.hashicorp.vault.conf=./conf/bootstrap-hashicorp-vault.conf
+
+# AWS KMS Sensitive Property Providers
+nifi.registry.bootstrap.protection.aws.kms.conf=./conf/bootstrap-aws.conf