Move new S3A encryption methods to a new class S3AEncryption

Use a single ITest class for encryption context with parameterized encryption algorithm and kms key

Author: raphaelazzolini

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java

@@ -18,7 +18,6 @@
 
 package org.apache.hadoop.fs.s3a;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
 import software.amazon.awssdk.awscore.exception.AwsServiceException;
 import software.amazon.awssdk.core.exception.AbortedException;
 import software.amazon.awssdk.core.exception.ApiCallAttemptTimeoutException;
@@ -28,7 +27,6 @@
 import software.amazon.awssdk.services.s3.model.S3Exception;
 import software.amazon.awssdk.services.s3.model.S3Object;
 
-import org.apache.commons.codec.binary.Base64;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
@@ -39,6 +37,7 @@
 import org.apache.hadoop.fs.PathFilter;
 import org.apache.hadoop.fs.PathIOException;
 import org.apache.hadoop.fs.RemoteIterator;
+import org.apache.hadoop.fs.s3a.impl.S3AEncryption;
 import org.apache.hadoop.util.functional.RemoteIterators;
 import org.apache.hadoop.fs.s3a.auth.delegation.EncryptionSecrets;
 import org.apache.hadoop.fs.s3a.impl.MultiObjectDeleteException;
@@ -64,7 +63,6 @@
 import java.lang.reflect.Modifier;
 import java.net.SocketTimeoutException;
 import java.net.URI;
-import java.nio.charset.StandardCharsets;
 import java.nio.file.AccessDeniedException;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -127,7 +125,6 @@ public final class S3AUtils {
       S3AEncryptionMethods.SSE_S3.getMethod()
           + " is enabled but an encryption key was set in "
           + Constants.S3_ENCRYPTION_KEY;
-
   public static final String EOF_MESSAGE_IN_XML_PARSER
       = "Failed to sanitize XML document destined for handler class";
 
@@ -1316,7 +1313,7 @@ static void patchSecurityCredentialProviders(Configuration conf) {
    * @throws IOException on any IO problem
    * @throws IllegalArgumentException bad arguments
    */
-  private static String lookupBucketSecret(
+  public static String lookupBucketSecret(
       String bucket,
       Configuration conf,
       String baseKey)
@@ -1406,79 +1403,6 @@ public static String getS3EncryptionKey(
     }
   }
 
-  /**
-   *  Get any SSE context, without propagating exceptions from
-   * JCEKs files.
-   * @param bucket bucket to query for
-   * @param conf configuration to examine
-   * @return the encryption context value or ""
-   * @throws IllegalArgumentException bad arguments.
-   */
-  public static String getS3EncryptionContext(
-      String bucket,
-      Configuration conf) {
-    try {
-      return getEncryptionContextValue(bucket, conf);
-    } catch (IOException e) {
-      // never going to happen, but to make sure, covert to
-      // runtime exception
-      throw new UncheckedIOException(e);
-    }
-  }
-
-  /**
-   * Get any SSE context from a configuration/credential provider.
-   * This includes converting the values to a base64-encoded UTF-8 string
-   * holding JSON with the encryption context key-value pairs
-   * @param bucket bucket to query for
-   * @param conf configuration to examine
-   * @param propagateExceptions should IO exceptions be rethrown?
-   * @return the Base64 encryption context or ""
-   * @throws IllegalArgumentException bad arguments.
-   * @throws IOException if propagateExceptions==true and reading a JCEKS file raised an IOE
-   */
-  public static String getS3EncryptionContextBase64Encoded(
-      String bucket,
-      Configuration conf,
-      boolean propagateExceptions) throws IOException {
-    try {
-      final String encryptionContextValue = getEncryptionContextValue(bucket, conf);
-      if (StringUtils.isBlank(encryptionContextValue)) {
-        return "";
-      }
-      final Map<String, String> encryptionContextMap = getTrimmedStringCollectionSplitByEquals(
-          encryptionContextValue);
-      if (encryptionContextMap.isEmpty()) {
-        return "";
-      }
-      final String encryptionContextJson = new ObjectMapper().writeValueAsString(
-          encryptionContextMap);
-      return Base64.encodeBase64String(encryptionContextJson.getBytes(StandardCharsets.UTF_8));
-    } catch (IOException e) {
-      if (propagateExceptions) {
-        throw e;
-      }
-      LOG.warn("Cannot retrieve {} for bucket {}",
-              S3_ENCRYPTION_CONTEXT, bucket, e);
-      return "";
-    }
-  }
-
-  private static String getEncryptionContextValue(String bucket, Configuration conf)
-      throws IOException {
-    // look up the per-bucket value of the encryption context
-    String encryptionContext = lookupBucketSecret(bucket, conf, S3_ENCRYPTION_CONTEXT);
-    if (encryptionContext == null) {
-      // look up the global value of the encryption context
-      encryptionContext = lookupPassword(null, conf, S3_ENCRYPTION_CONTEXT);
-    }
-    if (encryptionContext == null) {
-      // no encryption context, return ""
-      return "";
-    }
-    return encryptionContext;
-  }
-
   /**
    * Get the server-side encryption or client side encryption algorithm.
    * This includes validation of the configuration, checking the state of
@@ -1535,6 +1459,8 @@ public static EncryptionSecrets buildEncryptionSecrets(String bucket,
     int encryptionKeyLen =
         StringUtils.isBlank(encryptionKey) ? 0 : encryptionKey.length();
     String diagnostics = passwordDiagnostics(encryptionKey, "key");
+    String encryptionContext = S3AEncryption.getS3EncryptionContextBase64Encoded(bucket, conf,
+        encryptionMethod.requiresSecret());
     switch (encryptionMethod) {
     case SSE_C:
       LOG.debug("Using SSE-C with {}", diagnostics);
@@ -1570,9 +1496,6 @@ public static EncryptionSecrets buildEncryptionSecrets(String bucket,
       LOG.debug("Data is unencrypted");
       break;
     }
-
-    String encryptionContext = getS3EncryptionContextBase64Encoded(bucket, conf,
-        encryptionMethod.requiresSecret());
     return new EncryptionSecrets(encryptionMethod, encryptionKey, encryptionContext);
   }
 
@@ -1779,7 +1702,7 @@ public static Map<String, String> getTrimmedStringCollectionSplitByEquals(
    * @return property value as a <code>Map</code> of <code>String</code>s, or empty
    * <code>Map</code>.
    */
-  private static Map<String, String> getTrimmedStringCollectionSplitByEquals(
+  public static Map<String, String> getTrimmedStringCollectionSplitByEquals(
       final String valueString) {
     if (null == valueString) {
       return new HashMap<>();

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/auth/delegation/EncryptionSecretOperations.java

@@ -71,7 +71,6 @@ public static Optional<String> getSSEAwsKMSKey(final EncryptionSecrets secrets)
   public static Optional<String> getSSEAwsKMSEncryptionContext(final EncryptionSecrets secrets) {
     if ((secrets.getEncryptionMethod() == S3AEncryptionMethods.SSE_KMS
         || secrets.getEncryptionMethod() == S3AEncryptionMethods.DSSE_KMS)
-        && secrets.hasEncryptionKey()
         && secrets.hasEncryptionContext()) {
       return Optional.of(secrets.getEncryptionContext());
     } else {

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/RequestFactoryImpl.java

@@ -270,6 +270,8 @@ protected void copyEncryptionParameters(HeadObjectResponse srcom,
       LOG.debug("Propagating SSE-KMS settings from source {}",
           sourceKMSId);
       copyObjectRequestBuilder.ssekmsKeyId(sourceKMSId);
+      EncryptionSecretOperations.getSSEAwsKMSEncryptionContext(encryptionSecrets)
+          .ifPresent(copyObjectRequestBuilder::ssekmsEncryptionContext);
       return;
     }
 

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/S3AEncryption.java

@@ -0,0 +1,105 @@
+/*
+ * 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.hadoop.fs.s3a.impl;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.commons.codec.binary.Base64;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.s3a.S3AUtils;
+
+import static org.apache.hadoop.fs.s3a.Constants.S3_ENCRYPTION_CONTEXT;
+
+/**
+ * Utility methods for S3A encryption properties.
+ */
+public final class S3AEncryption {
+
+  private static final Logger LOG = LoggerFactory.getLogger(S3AEncryption.class);
+
+  private S3AEncryption() {
+  }
+
+  /**
+   * Get any SSE context from a configuration/credential provider.
+   * @param bucket bucket to query for
+   * @param conf configuration to examine
+   * @return the encryption context value or ""
+   * @throws IllegalArgumentException bad arguments.
+   */
+  public static String getS3EncryptionContext(String bucket, Configuration conf)
+      throws IOException {
+    // look up the per-bucket value of the encryption context
+    String encryptionContext = S3AUtils.lookupBucketSecret(bucket, conf, S3_ENCRYPTION_CONTEXT);
+    if (encryptionContext == null) {
+      // look up the global value of the encryption context
+      encryptionContext = S3AUtils.lookupPassword(null, conf, S3_ENCRYPTION_CONTEXT);
+    }
+    if (encryptionContext == null) {
+      // no encryption context, return ""
+      return "";
+    }
+    return encryptionContext;
+  }
+
+  /**
+   * Get any SSE context from a configuration/credential provider.
+   * This includes converting the values to a base64-encoded UTF-8 string
+   * holding JSON with the encryption context key-value pairs
+   * @param bucket bucket to query for
+   * @param conf configuration to examine
+   * @param propagateExceptions should IO exceptions be rethrown?
+   * @return the Base64 encryption context or ""
+   * @throws IllegalArgumentException bad arguments.
+   * @throws IOException if propagateExceptions==true and reading a JCEKS file raised an IOE
+   */
+  public static String getS3EncryptionContextBase64Encoded(
+      String bucket,
+      Configuration conf,
+      boolean propagateExceptions) throws IOException {
+    try {
+      final String encryptionContextValue = getS3EncryptionContext(bucket, conf);
+      if (StringUtils.isBlank(encryptionContextValue)) {
+        return "";
+      }
+      final Map<String, String> encryptionContextMap = S3AUtils
+          .getTrimmedStringCollectionSplitByEquals(encryptionContextValue);
+      if (encryptionContextMap.isEmpty()) {
+        return "";
+      }
+      final String encryptionContextJson = new ObjectMapper().writeValueAsString(
+          encryptionContextMap);
+      return Base64.encodeBase64String(encryptionContextJson.getBytes(StandardCharsets.UTF_8));
+    } catch (IOException e) {
+      if (propagateExceptions) {
+        throw e;
+      }
+      LOG.warn("Cannot retrieve {} for bucket {}",
+          S3_ENCRYPTION_CONTEXT, bucket, e);
+      return "";
+    }
+  }
+}