HADOOP-17912. ABFS: Support for Encryption Context

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AbfsConfiguration.java

@@ -21,6 +21,7 @@
 import java.io.IOException;
 import java.lang.reflect.Field;
 
+import org.apache.hadoop.fs.azurebfs.extensions.EncryptionContextProvider;
 import org.apache.hadoop.thirdparty.com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.thirdparty.com.google.common.base.Preconditions;
 
@@ -906,6 +907,36 @@ public SASTokenProvider getSASTokenProvider() throws AzureBlobFileSystemExceptio
     }
   }
 
+  public EncryptionContextProvider initializeEncryptionContextProvider() {
+
+    try {
+      String configKey = FS_AZURE_ENCRYPTION_CONTEXT_PROVIDER_TYPE;
+      if (get(configKey) == null) {
+        return null;
+      }
+      Class<? extends EncryptionContextProvider> encryptionContextClass =
+          getAccountSpecificClass(configKey, null, EncryptionContextProvider.class);
+      if (encryptionContextClass == null) {
+        encryptionContextClass = getAccountAgnosticClass(configKey, null,
+            EncryptionContextProvider.class);
+      }
+      Preconditions.checkArgument(encryptionContextClass != null,
+          String.format("The configuration value for %s is invalid.", configKey));
+
+      EncryptionContextProvider encryptionContextProvider =
+          ReflectionUtils.newInstance(encryptionContextClass, rawConfig);
+      Preconditions.checkArgument(encryptionContextProvider != null,
+          String.format("Failed to initialize %s", encryptionContextClass));
+
+      LOG.trace("Initializing {}", encryptionContextClass.getName());
+      LOG.trace("{} init complete", encryptionContextClass.getName());
+      return encryptionContextProvider;
+    } catch (Exception e) {
+      throw new IllegalArgumentException("Unable to load encryption context provider class: ", e);
+    }
+
+  }
+
   public int getReadAheadRange() {
     return this.readAheadRange;
   }
@@ -1009,7 +1040,7 @@ public boolean enableAbfsListIterator() {
   }
 
   public String getClientProvidedEncryptionKey() {
-    String accSpecEncKey = accountConf(FS_AZURE_CLIENT_PROVIDED_ENCRYPTION_KEY);
+    String accSpecEncKey = accountConf(FS_AZURE_ENCRYPTION_CLIENT_PROVIDED_KEY);
     return rawConfig.get(accSpecEncKey, null);
   }
 

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystemStore.java

@@ -54,6 +54,9 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.hadoop.fs.azurebfs.extensions.EncryptionContextProvider;
+import org.apache.hadoop.fs.azurebfs.security.EncryptionAdapter;
+import org.apache.hadoop.fs.azurebfs.utils.EncryptionType;
 import org.apache.hadoop.thirdparty.com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.thirdparty.com.google.common.base.Preconditions;
 import org.apache.hadoop.thirdparty.com.google.common.base.Strings;
@@ -460,7 +463,7 @@ public void setFilesystemProperties(
   }
 
   public Hashtable<String, String> getPathStatus(final Path path,
-      TracingContext tracingContext) throws AzureBlobFileSystemException {
+      TracingContext tracingContext) throws IOException {
     try (AbfsPerfInfo perfInfo = startTracking("getPathStatus", "getPathStatus")){
       LOG.debug("getPathStatus for filesystem: {} path: {}",
               client.getFileSystem(),
@@ -483,7 +486,7 @@ public Hashtable<String, String> getPathStatus(final Path path,
 
   public void setPathProperties(final Path path,
       final Hashtable<String, String> properties, TracingContext tracingContext)
-      throws AzureBlobFileSystemException {
+      throws IOException {
     try (AbfsPerfInfo perfInfo = startTracking("setPathProperties", "setPathProperties")){
       LOG.debug("setFilesystemProperties for filesystem: {} path: {} with properties: {}",
               client.getFileSystem(),
@@ -554,13 +557,16 @@ public OutputStream createFile(final Path path,
         triggerConditionalCreateOverwrite = true;
       }
 
+      EncryptionAdapter encryptionAdapter = new EncryptionAdapter(
+          client.getEncryptionContextProvider(), getRelativePath(path));
       AbfsRestOperation op;
       if (triggerConditionalCreateOverwrite) {
         op = conditionalCreateOverwriteFile(relativePath,
             statistics,
             isNamespaceEnabled ? getOctalNotation(permission) : null,
             isNamespaceEnabled ? getOctalNotation(umask) : null,
             isAppendBlob,
+            encryptionAdapter,
             tracingContext
         );
 
@@ -571,6 +577,7 @@ public OutputStream createFile(final Path path,
             isNamespaceEnabled ? getOctalNotation(umask) : null,
             isAppendBlob,
             null,
+            encryptionAdapter,
             tracingContext);
 
       }
@@ -586,6 +593,7 @@ public OutputStream createFile(final Path path,
               statistics,
               relativePath,
               0,
+              encryptionAdapter,
               tracingContext));
     }
   }
@@ -606,15 +614,16 @@ private AbfsRestOperation conditionalCreateOverwriteFile(final String relativePa
       final String permission,
       final String umask,
       final boolean isAppendBlob,
-      TracingContext tracingContext) throws AzureBlobFileSystemException {
+      EncryptionAdapter encryptionAdapter,
+      TracingContext tracingContext) throws IOException {
     AbfsRestOperation op;
 
     try {
       // Trigger a create with overwrite=false first so that eTag fetch can be
       // avoided for cases when no pre-existing file is present (major portion
       // of create file traffic falls into the case of no pre-existing file).
       op = client.createPath(relativePath, true, false, permission, umask,
-          isAppendBlob, null, tracingContext);
+          isAppendBlob, null, encryptionAdapter, tracingContext);
 
     } catch (AbfsRestOperationException e) {
       if (e.getStatusCode() == HttpURLConnection.HTTP_CONFLICT) {
@@ -639,7 +648,7 @@ private AbfsRestOperation conditionalCreateOverwriteFile(final String relativePa
         try {
           // overwrite only if eTag matches with the file properties fetched befpre
           op = client.createPath(relativePath, true, true, permission, umask,
-              isAppendBlob, eTag, tracingContext);
+              isAppendBlob, eTag, encryptionAdapter, tracingContext);
         } catch (AbfsRestOperationException ex) {
           if (ex.getStatusCode() == HttpURLConnection.HTTP_PRECON_FAILED) {
             // Is a parallel access case, as file with eTag was just queried
@@ -682,6 +691,7 @@ private AbfsOutputStreamContext populateAbfsOutputStreamContext(
       FileSystem.Statistics statistics,
       String path,
       long position,
+      EncryptionAdapter encryptionAdapter,
       TracingContext tracingContext) {
     int bufferSize = abfsConfiguration.getWriteBufferSize();
     if (isAppendBlob && bufferSize > FileSystemConfigurations.APPENDBLOB_MAX_WRITE_BUFFER_SIZE) {
@@ -697,6 +707,7 @@ private AbfsOutputStreamContext populateAbfsOutputStreamContext(
             .withWriteMaxConcurrentRequestCount(abfsConfiguration.getWriteMaxConcurrentRequestCount())
             .withMaxWriteRequestsToQueue(abfsConfiguration.getMaxWriteRequestsToQueue())
             .withLease(lease)
+            .withEncryptionAdapter(encryptionAdapter)
             .withBlockFactory(blockFactory)
             .withBlockOutputActiveBlocks(blockOutputActiveBlocks)
             .withClient(client)
@@ -711,7 +722,7 @@ private AbfsOutputStreamContext populateAbfsOutputStreamContext(
 
   public void createDirectory(final Path path, final FsPermission permission,
       final FsPermission umask, TracingContext tracingContext)
-      throws AzureBlobFileSystemException {
+      throws IOException {
     try (AbfsPerfInfo perfInfo = startTracking("createDirectory", "createPath")) {
       boolean isNamespaceEnabled = getIsNamespaceEnabled(tracingContext);
       LOG.debug("createDirectory filesystem: {} path: {} permission: {} umask: {} isNamespaceEnabled: {}",
@@ -726,8 +737,8 @@ public void createDirectory(final Path path, final FsPermission permission,
       final AbfsRestOperation op = client.createPath(getRelativePath(path),
           false, overwrite,
               isNamespaceEnabled ? getOctalNotation(permission) : null,
-              isNamespaceEnabled ? getOctalNotation(umask) : null, false, null,
-              tracingContext);
+              isNamespaceEnabled ? getOctalNotation(umask) : null, false,
+          null, null, tracingContext);
       perfInfo.registerResult(op.getResult()).registerSuccess(true);
     }
   }
@@ -753,7 +764,9 @@ public AbfsInputStream openFileForRead(Path path,
       String relativePath = getRelativePath(path);
       String resourceType, eTag;
       long contentLength;
-      if (fileStatus instanceof VersionedFileStatus) {
+      EncryptionAdapter encryptionAdapter = null;
+      if (fileStatus instanceof VersionedFileStatus
+          && client.getEncryptionType() != EncryptionType.ENCRYPTION_CONTEXT) {
         path = path.makeQualified(this.uri, path);
         Preconditions.checkArgument(fileStatus.getPath().equals(path),
             String.format(
@@ -764,9 +777,8 @@ public AbfsInputStream openFileForRead(Path path,
         eTag = ((VersionedFileStatus) fileStatus).getVersion();
       } else {
         if (fileStatus != null) {
-          LOG.warn(
-              "Fallback to getPathStatus REST call as provided filestatus "
-                  + "is not of type VersionedFileStatus");
+          LOG.debug("Fallback to getPathStatus REST call as provided fileStatus "
+              + "is not of type VersionedFileStatus, or file is encrypted");
         }
         AbfsHttpOperation op = client.getPathStatus(relativePath, false,
             tracingContext).getResult();
@@ -775,6 +787,12 @@ public AbfsInputStream openFileForRead(Path path,
         contentLength = Long.parseLong(
             op.getResponseHeader(HttpHeaderConfigurations.CONTENT_LENGTH));
         eTag = op.getResponseHeader(HttpHeaderConfigurations.ETAG);
+        if (client.getEncryptionType() == EncryptionType.ENCRYPTION_CONTEXT) {
+          encryptionAdapter = new EncryptionAdapter(
+              client.getEncryptionContextProvider(), getRelativePath(path),
+              op.getResponseHeader(HttpHeaderConfigurations.X_MS_ENCRYPTION_CONTEXT)
+                  .getBytes(StandardCharsets.UTF_8));
+        }
       }
 
       if (parseIsDirectory(resourceType)) {
@@ -790,13 +808,13 @@ public AbfsInputStream openFileForRead(Path path,
       // Add statistics for InputStream
       return new AbfsInputStream(client, statistics, relativePath,
           contentLength, populateAbfsInputStreamContext(
-          parameters.map(OpenFileParameters::getOptions)),
+          parameters.map(OpenFileParameters::getOptions), encryptionAdapter),
           eTag, tracingContext);
     }
   }
 
   private AbfsInputStreamContext populateAbfsInputStreamContext(
-      Optional<Configuration> options) {
+      Optional<Configuration> options, EncryptionAdapter encryptionAdapter) {
     boolean bufferedPreadDisabled = options
         .map(c -> c.getBoolean(FS_AZURE_BUFFERED_PREAD_DISABLE, false))
         .orElse(false);
@@ -812,6 +830,7 @@ private AbfsInputStreamContext populateAbfsInputStreamContext(
                 abfsConfiguration.shouldReadBufferSizeAlways())
             .withReadAheadBlockSize(abfsConfiguration.getReadAheadBlockSize())
             .withBufferedPreadDisabled(bufferedPreadDisabled)
+            .withEncryptionAdapter(encryptionAdapter)
             .build();
   }
 
@@ -851,6 +870,12 @@ public OutputStream openFileForWrite(final Path path,
       }
 
       AbfsLease lease = maybeCreateLease(relativePath, tracingContext);
+      byte[] encryptionContext = op.getResult()
+          .getResponseHeader(HttpHeaderConfigurations.X_MS_ENCRYPTION_CONTEXT)
+          .getBytes(StandardCharsets.UTF_8);
+      EncryptionAdapter encryptionAdapter = new EncryptionAdapter(
+          client.getEncryptionContextProvider(), getRelativePath(path),
+          encryptionContext);
 
       return new AbfsOutputStream(
           populateAbfsOutputStreamContext(
@@ -860,6 +885,7 @@ public OutputStream openFileForWrite(final Path path,
               statistics,
               relativePath,
               offset,
+              encryptionAdapter,
               tracingContext));
     }
   }
@@ -877,8 +903,8 @@ public void breakLease(final Path path, final TracingContext tracingContext) thr
     client.breakLease(getRelativePath(path), tracingContext);
   }
 
-  public void rename(final Path source, final Path destination, TracingContext tracingContext) throws
-          AzureBlobFileSystemException {
+  public void rename(final Path source, final Path destination, TracingContext tracingContext)
+      throws IOException {
     final Instant startAggregate = abfsPerfTracker.getLatencyInstant();
     long countAggregate = 0;
     boolean shouldContinue;
@@ -1590,16 +1616,38 @@ private void initializeClient(URI uri, String fileSystemName,
             abfsConfiguration.getRawConfiguration());
     }
 
+    // Encryption setup
+    EncryptionType encryptionType = EncryptionType.NONE;
+    EncryptionContextProvider encryptionContextProvider = null;
+    if (isSecure) {
+      encryptionContextProvider =
+          abfsConfiguration.initializeEncryptionContextProvider();
+      if (encryptionContextProvider != null) {
+        if (abfsConfiguration.getClientProvidedEncryptionKey() != null) {
+          throw new IOException(
+              "Both global key and encryption context are set, only one allowed");
+        }
+        encryptionContextProvider.initialize(
+            abfsConfiguration.getRawConfiguration(), accountName,
+            fileSystemName);
+        encryptionType = EncryptionType.ENCRYPTION_CONTEXT;
+      } else if (abfsConfiguration.getClientProvidedEncryptionKey() != null) {
+        encryptionType = EncryptionType.GLOBAL_KEY;
+      }
+    }
+
     LOG.trace("Initializing AbfsClient for {}", baseUrl);
     if (tokenProvider != null) {
       this.client = new AbfsClient(baseUrl, creds, abfsConfiguration,
-          tokenProvider,
+          tokenProvider, encryptionContextProvider,
           populateAbfsClientContext());
     } else {
       this.client = new AbfsClient(baseUrl, creds, abfsConfiguration,
-          sasTokenProvider,
+          sasTokenProvider, encryptionContextProvider,
           populateAbfsClientContext());
     }
+    client.setEncryptionType(encryptionType);
+
     LOG.trace("AbfsClient init complete");
   }
 
@@ -1622,7 +1670,7 @@ private String getOctalNotation(FsPermission fsPermission) {
     return String.format(AbfsHttpConstants.PERMISSION_FORMAT, fsPermission.toOctal());
   }
 
-  private String getRelativePath(final Path path) {
+  public String getRelativePath(final Path path) {
     Preconditions.checkNotNull(path, "path");
     return path.toUri().getPath();
   }
@@ -1640,7 +1688,8 @@ private boolean parseIsDirectory(final String resourceType) {
         && resourceType.equalsIgnoreCase(AbfsHttpConstants.DIRECTORY);
   }
 
-  private String convertXmsPropertiesToCommaSeparatedString(final Hashtable<String, String> properties) throws
+  public String convertXmsPropertiesToCommaSeparatedString(final Hashtable<String,
+      String> properties) throws
           CharacterCodingException {
     StringBuilder commaSeparatedProperties = new StringBuilder();
 

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/ConfigurationKeys.java

@@ -189,7 +189,9 @@ public final class ConfigurationKeys {
   /** Setting this true will make the driver use it's own RemoteIterator implementation */
   public static final String FS_AZURE_ENABLE_ABFS_LIST_ITERATOR = "fs.azure.enable.abfslistiterator";
   /** Server side encryption key */
-  public static final String FS_AZURE_CLIENT_PROVIDED_ENCRYPTION_KEY = "fs.azure.client-provided-encryption-key";
+  public static final String FS_AZURE_ENCRYPTION_CLIENT_PROVIDED_KEY = "fs.azure.encryption.client-provided-key";
+  /** Custom EncryptionContextProvider type */
+  public static final String FS_AZURE_ENCRYPTION_CONTEXT_PROVIDER_TYPE = "fs.azure.encryption.context.provider.type";
 
   /** End point of ABFS account: {@value}. */
   public static final String AZURE_ABFS_ENDPOINT = "fs.azure.abfs.endpoint";

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/HttpHeaderConfigurations.java

@@ -65,6 +65,7 @@ public final class HttpHeaderConfigurations {
   public static final String X_MS_ENCRYPTION_ALGORITHM = "x-ms-encryption-algorithm";
   public static final String X_MS_REQUEST_SERVER_ENCRYPTED = "x-ms-request-server-encrypted";
   public static final String X_MS_SERVER_ENCRYPTED = "x-ms-server-encrypted";
+  public static final String X_MS_ENCRYPTION_CONTEXT = "x-ms-encryption-context";
   public static final String X_MS_LEASE_ACTION = "x-ms-lease-action";
   public static final String X_MS_LEASE_DURATION = "x-ms-lease-duration";
   public static final String X_MS_LEASE_ID = "x-ms-lease-id";