NIFI-8501: Setup Azure blob client side encryption

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

@@ -22,6 +22,7 @@
     <properties>
         <azure-eventhubs.version>3.2.1</azure-eventhubs.version>
         <azure-eventhubs-eph.version>3.2.1</azure-eventhubs-eph.version>
+        <azure-keyvault.version>1.2.4</azure-keyvault.version>
     </properties>
     <dependencies>
         <dependency>
@@ -83,6 +84,11 @@
             <artifactId>azure-eventhubs</artifactId>
             <version>${azure-eventhubs.version}</version>
         </dependency>
+        <dependency>
+            <groupId>com.microsoft.azure</groupId>
+            <artifactId>azure-keyvault</artifactId>
+            <version>${azure-keyvault.version}</version>
+        </dependency>
         <dependency>
             <groupId>com.microsoft.azure</groupId>
             <artifactId>azure-eventhubs-eph</artifactId>

nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/AbstractAzureBlobProcessor.java

@@ -16,13 +16,21 @@
  */
 package org.apache.nifi.processors.azure;
 
+import com.microsoft.azure.keyvault.cryptography.SymmetricKey;
+import com.microsoft.azure.storage.blob.BlobEncryptionPolicy;
+import com.microsoft.azure.storage.blob.BlobRequestOptions;
+import org.apache.commons.codec.DecoderException;
+import org.apache.commons.codec.binary.Hex;
 import org.apache.nifi.components.PropertyDescriptor;
 import org.apache.nifi.components.ValidationContext;
 import org.apache.nifi.components.ValidationResult;
 import org.apache.nifi.expression.ExpressionLanguageScope;
 import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
 import org.apache.nifi.processor.Relationship;
 import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.azure.storage.utils.AzureBlobClientSideEncryptionMethod;
+import org.apache.nifi.processors.azure.storage.utils.AzureBlobClientSideEncryptionUtils;
 import org.apache.nifi.processors.azure.storage.utils.AzureStorageUtils;
 
 import java.util.Arrays;
@@ -85,4 +93,24 @@ protected Collection<ValidationResult> customValidate(ValidationContext validati
     public Set<Relationship> getRelationships() {
         return RELATIONSHIPS;
     }
+
+    protected BlobRequestOptions createBlobRequestOptions(ProcessContext context) throws DecoderException {
+        final String cseKeyTypeValue = context.getProperty(AzureBlobClientSideEncryptionUtils.CSE_KEY_TYPE).getValue();
+        final AzureBlobClientSideEncryptionMethod cseKeyType = AzureBlobClientSideEncryptionMethod.valueOf(cseKeyTypeValue);
+
+        final String cseKeyId = context.getProperty(AzureBlobClientSideEncryptionUtils.CSE_KEY_ID).getValue();
+
+        final String cseSymmetricKeyHex = context.getProperty(AzureBlobClientSideEncryptionUtils.CSE_SYMMETRIC_KEY_HEX).getValue();
+
+        BlobRequestOptions blobRequestOptions = new BlobRequestOptions();
+
+        if (cseKeyType == AzureBlobClientSideEncryptionMethod.SYMMETRIC) {
+            byte[] keyBytes = Hex.decodeHex(cseSymmetricKeyHex.toCharArray());
+            SymmetricKey key = new SymmetricKey(cseKeyId, keyBytes);
+            BlobEncryptionPolicy policy = new BlobEncryptionPolicy(key, null);
+            blobRequestOptions.setEncryptionPolicy(policy);
+        }
+
+        return blobRequestOptions;
+    }
 }

nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/FetchAzureBlobStorage.java

@@ -23,10 +23,12 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Collection;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
 import com.microsoft.azure.storage.OperationContext;
+import org.apache.commons.codec.DecoderException;
 import org.apache.nifi.annotation.behavior.InputRequirement;
 import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
 import org.apache.nifi.annotation.behavior.WritesAttribute;
@@ -35,6 +37,8 @@
 import org.apache.nifi.annotation.documentation.SeeAlso;
 import org.apache.nifi.annotation.documentation.Tags;
 import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
 import org.apache.nifi.expression.ExpressionLanguageScope;
 import org.apache.nifi.flowfile.FlowFile;
 import org.apache.nifi.processor.DataUnit;
@@ -43,12 +47,14 @@
 import org.apache.nifi.processor.exception.ProcessException;
 import org.apache.nifi.processor.util.StandardValidators;
 import org.apache.nifi.processors.azure.AbstractAzureBlobProcessor;
+import org.apache.nifi.processors.azure.storage.utils.AzureBlobClientSideEncryptionUtils;
 import org.apache.nifi.processors.azure.storage.utils.AzureStorageUtils;
 
 import com.microsoft.azure.storage.StorageException;
 import com.microsoft.azure.storage.blob.CloudBlob;
 import com.microsoft.azure.storage.blob.CloudBlobClient;
 import com.microsoft.azure.storage.blob.CloudBlobContainer;
+import com.microsoft.azure.storage.blob.BlobRequestOptions;
 
 @Tags({ "azure", "microsoft", "cloud", "storage", "blob" })
 @CapabilityDescription("Retrieves contents of an Azure Storage Blob, writing the contents to the content of the FlowFile")
@@ -79,11 +85,21 @@ public class FetchAzureBlobStorage extends AbstractAzureBlobProcessor {
             .required(false)
             .build();
 
+    @Override
+    protected Collection<ValidationResult> customValidate(ValidationContext validationContext) {
+        final List<ValidationResult> results = new ArrayList<>(super.customValidate(validationContext));
+        results.addAll(AzureBlobClientSideEncryptionUtils.validateClientSideEncryptionProperties(validationContext));
+        return results;
+    }
+
     @Override
     public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
-        List<PropertyDescriptor> properties = new ArrayList<PropertyDescriptor>(super.getSupportedPropertyDescriptors());
+        List<PropertyDescriptor> properties = new ArrayList<>(super.getSupportedPropertyDescriptors());
         properties.add(RANGE_START);
         properties.add(RANGE_LENGTH);
+        properties.add(AzureBlobClientSideEncryptionUtils.CSE_KEY_TYPE);
+        properties.add(AzureBlobClientSideEncryptionUtils.CSE_KEY_ID);
+        properties.add(AzureBlobClientSideEncryptionUtils.CSE_SYMMETRIC_KEY_HEX);
         return properties;
     }
 
@@ -112,11 +128,13 @@ public void onTrigger(ProcessContext context, ProcessSession session) throws Pro
             final Map<String, String> attributes = new HashMap<>();
             final CloudBlob blob = container.getBlockBlobReference(blobPath);
 
+            BlobRequestOptions blobRequestOptions = createBlobRequestOptions(context);
+
             // TODO - we may be able do fancier things with ranges and
             // distribution of download over threads, investigate
             flowFile = session.write(flowFile, os -> {
                 try {
-                    blob.downloadRange(rangeStart, rangeLength, os, null, null, operationContext);
+                    blob.downloadRange(rangeStart, rangeLength, os, null, blobRequestOptions, operationContext);
                 } catch (StorageException e) {
                     storedException.set(e);
                     throw new IOException(e);
@@ -133,7 +151,7 @@ public void onTrigger(ProcessContext context, ProcessSession session) throws Pro
             session.transfer(flowFile, REL_SUCCESS);
             final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
             session.getProvenanceReporter().fetch(flowFile, blob.getSnapshotQualifiedUri().toString(), transferMillis);
-        } catch (IllegalArgumentException | URISyntaxException | StorageException | ProcessException e) {
+        } catch (IllegalArgumentException | URISyntaxException | StorageException | ProcessException | DecoderException e) {
             if (e instanceof ProcessException && storedException.get() == null) {
                 throw (ProcessException) e;
             } else {

nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/PutAzureBlobStorage.java

@@ -26,11 +26,13 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Collection;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.microsoft.azure.storage.OperationContext;
+import org.apache.commons.codec.DecoderException;
 import org.apache.nifi.annotation.behavior.InputRequirement;
 import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
 import org.apache.nifi.annotation.behavior.WritesAttribute;
@@ -39,20 +41,24 @@
 import org.apache.nifi.annotation.documentation.SeeAlso;
 import org.apache.nifi.annotation.documentation.Tags;
 import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
 import org.apache.nifi.expression.ExpressionLanguageScope;
 import org.apache.nifi.flowfile.FlowFile;
 import org.apache.nifi.processor.ProcessContext;
 import org.apache.nifi.processor.ProcessSession;
 import org.apache.nifi.processor.exception.ProcessException;
 import org.apache.nifi.processor.util.StandardValidators;
 import org.apache.nifi.processors.azure.AbstractAzureBlobProcessor;
+import org.apache.nifi.processors.azure.storage.utils.AzureBlobClientSideEncryptionUtils;
 import org.apache.nifi.processors.azure.storage.utils.AzureStorageUtils;
 
 import com.microsoft.azure.storage.StorageException;
 import com.microsoft.azure.storage.blob.BlobProperties;
 import com.microsoft.azure.storage.blob.CloudBlob;
 import com.microsoft.azure.storage.blob.CloudBlobClient;
 import com.microsoft.azure.storage.blob.CloudBlobContainer;
+import com.microsoft.azure.storage.blob.BlobRequestOptions;
 
 @Tags({ "azure", "microsoft", "cloud", "storage", "blob" })
 @SeeAlso({ ListAzureBlobStorage.class, FetchAzureBlobStorage.class, DeleteAzureBlobStorage.class })
@@ -88,12 +94,22 @@ public class PutAzureBlobStorage extends AbstractAzureBlobProcessor {
                   "will fail if the container does not exist.")
             .build();
 
+    @Override
+    protected Collection<ValidationResult> customValidate(ValidationContext validationContext) {
+        final List<ValidationResult> results = new ArrayList<>(super.customValidate(validationContext));
+        results.addAll(AzureBlobClientSideEncryptionUtils.validateClientSideEncryptionProperties(validationContext));
+        return results;
+    }
+
     @Override
     public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
         List<PropertyDescriptor> properties = new ArrayList<>(super.getSupportedPropertyDescriptors());
         properties.remove(BLOB);
         properties.add(BLOB_NAME);
         properties.add(CREATE_CONTAINER);
+        properties.add(AzureBlobClientSideEncryptionUtils.CSE_KEY_TYPE);
+        properties.add(AzureBlobClientSideEncryptionUtils.CSE_KEY_ID);
+        properties.add(AzureBlobClientSideEncryptionUtils.CSE_SYMMETRIC_KEY_HEX);
         return properties;
     }
 
@@ -124,6 +140,8 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
             final OperationContext operationContext = new OperationContext();
             AzureStorageUtils.setProxy(operationContext, context);
 
+            BlobRequestOptions blobRequestOptions = createBlobRequestOptions(context);
+
             final Map<String, String> attributes = new HashMap<>();
             long length = flowFile.getSize();
             session.read(flowFile, rawIn -> {
@@ -142,7 +160,7 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
                 }
 
                 try {
-                    uploadBlob(blob, operationContext, in);
+                    uploadBlob(blob, operationContext, blobRequestOptions, in);
                     BlobProperties properties = blob.getProperties();
                     attributes.put("azure.container", containerName);
                     attributes.put("azure.primaryUri", blob.getSnapshotQualifiedUri().toString());
@@ -163,7 +181,7 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
             final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
             session.getProvenanceReporter().send(flowFile, blob.getSnapshotQualifiedUri().toString(), transferMillis);
 
-        } catch (IllegalArgumentException | URISyntaxException | StorageException | ProcessException e) {
+        } catch (IllegalArgumentException | URISyntaxException | StorageException | ProcessException | DecoderException e) {
             if (e instanceof ProcessException && storedException.get() == null) {
                 throw (ProcessException) e;
             } else {
@@ -177,8 +195,8 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
     }
 
     @VisibleForTesting
-    void uploadBlob(CloudBlob blob, OperationContext operationContext, InputStream in) throws StorageException, IOException {
-        blob.upload(in, -1, null, null, operationContext);
+    void uploadBlob(CloudBlob blob, OperationContext operationContext, BlobRequestOptions blobRequestOptions, InputStream in) throws StorageException, IOException {
+        blob.upload(in, -1, null, blobRequestOptions, operationContext);
     }
 
     // Used to help force Azure Blob SDK to write in blocks

nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/utils/AzureBlobClientSideEncryptionMethod.java

@@ -0,0 +1,48 @@
+/*
+ * 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.processors.azure.storage.utils;
+
+/**
+ * Enumeration capturing essential information about the various client-side
+ * encryption methods supported by Azure
+ */
+public enum AzureBlobClientSideEncryptionMethod {
+
+    NONE("None", "The blobs sent to Azure are not encrypted."),
+    SYMMETRIC("Symmetric","The blobs sent to Azure are encrypted using a symmetric algorithm.");
+
+    private final String cseName;
+    private final String description;
+
+    AzureBlobClientSideEncryptionMethod(String cseName, String description) {
+        this.cseName = cseName;
+        this.description = description;
+    }
+
+    public String getCseName() {
+        return cseName;
+    }
+
+    public String getDescription() {
+        return description;
+    }
+
+    @Override
+    public String toString() {
+        return description;
+    }
+}