HADOOP-16794. S3A reverts KMS encryption to the bucket's default KMS …

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

@@ -2489,7 +2489,7 @@ UploadResult executePut(PutObjectRequest putObjectRequest,
    * Wait for an upload to complete.
    * If the waiting for completion is interrupted, the upload will be
    * aborted before an {@code InterruptedIOException} is thrown.
-   * @param upload upload to wait for
+   * @param uploadInfo upload to wait for
    * @param key destination key
    * @return the upload result
    * @throws InterruptedIOException if the blocking was interrupted.
@@ -2592,7 +2592,7 @@ private void copyFile(String srcKey, String dstKey, long size)
           setOptionalObjectMetadata(dstom);
           CopyObjectRequest copyObjectRequest =
               new CopyObjectRequest(bucket, srcKey, bucket, dstKey);
-          setOptionalCopyObjectRequestParameters(copyObjectRequest);
+          setOptionalCopyObjectRequestParameters(srcom, copyObjectRequest);
           copyObjectRequest.setCannedAccessControlList(cannedACL);
           copyObjectRequest.setNewObjectMetadata(dstom);
           Copy copy = transfers.copy(copyObjectRequest);
@@ -2608,6 +2608,49 @@ private void copyFile(String srcKey, String dstKey, long size)
         });
   }
 
+  /**
+   * Propagate encryption parameters from source file if set else use the
+   * current filesystem encryption settings.
+   * @param srcom source object meta.
+   * @param copyObjectRequest copy object request body.
+   */
+  private void setOptionalCopyObjectRequestParameters(
+          ObjectMetadata srcom,
+          CopyObjectRequest copyObjectRequest) {
+    String sourceKMSId = srcom.getSSEAwsKmsKeyId();
+    if (isNotEmpty(sourceKMSId)) {
+      // source KMS ID is propagated
+      LOG.debug("Propagating SSE-KMS settings from source {}",
+          sourceKMSId);
+      copyObjectRequest.setSSEAwsKeyManagementParams(
+              new SSEAwsKeyManagementParams(sourceKMSId));
+    }
+    switch(getServerSideEncryptionAlgorithm()) {
+    /**
+     * Overriding with client encryption settings.
+     */
+    case SSE_C:
+      if (isNotBlank(getServerSideEncryptionKey(bucket, getConf()))) {
+        //at the moment, only supports copy using the same key
+        SSECustomerKey customerKey = generateSSECustomerKey();
+        copyObjectRequest.setSourceSSECustomerKey(customerKey);
+        copyObjectRequest.setDestinationSSECustomerKey(customerKey);
+      }
+      break;
+    case SSE_KMS:
+      copyObjectRequest.setSSEAwsKeyManagementParams(
+              generateSSEAwsKeyParams()
+      );
+      break;
+    default:
+    }
+  }
+
+  /**
+   * Set the optional parameters when initiating the request (encryption,
+   * headers, storage, etc).
+   * @param req request to patch.
+   */
   protected void setOptionalMultipartUploadRequestParameters(
       InitiateMultipartUploadRequest req) {
     switch (serverSideEncryptionAlgorithm) {
@@ -2657,26 +2700,6 @@ InitiateMultipartUploadResult initiateMultipartUpload(
     return getAmazonS3Client().initiateMultipartUpload(request);
   }
 
-  protected void setOptionalCopyObjectRequestParameters(
-      CopyObjectRequest copyObjectRequest) throws IOException {
-    switch (serverSideEncryptionAlgorithm) {
-    case SSE_KMS:
-      copyObjectRequest.setSSEAwsKeyManagementParams(
-          generateSSEAwsKeyParams()
-      );
-      break;
-    case SSE_C:
-      if (isNotBlank(getServerSideEncryptionKey(bucket, getConf()))) {
-        //at the moment, only supports copy using the same key
-        SSECustomerKey customerKey = generateSSECustomerKey();
-        copyObjectRequest.setSourceSSECustomerKey(customerKey);
-        copyObjectRequest.setDestinationSSECustomerKey(customerKey);
-      }
-      break;
-    default:
-    }
-  }
-
   private void setOptionalPutRequestParameters(PutObjectRequest request) {
     switch (serverSideEncryptionAlgorithm) {
     case SSE_KMS:

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/AbstractS3ATestBase.java

@@ -39,7 +39,6 @@
  */
 public abstract class AbstractS3ATestBase extends AbstractFSContractTestBase
     implements S3ATestConstants {
-
   protected static final Logger LOG =
       LoggerFactory.getLogger(AbstractS3ATestBase.class);
 

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/AbstractTestS3AEncryption.java

@@ -21,15 +21,19 @@
 import com.amazonaws.services.s3.model.ObjectMetadata;
 import org.apache.commons.codec.digest.DigestUtils;
 import org.apache.commons.net.util.Base64;
+import java.io.IOException;
+
+import org.junit.Test;
+
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.contract.ContractTestUtils;
-import org.junit.Test;
 
-import java.io.IOException;
 
 import static org.apache.hadoop.fs.contract.ContractTestUtils.*;
-import static org.apache.hadoop.fs.s3a.S3ATestUtils.*;
+import static org.apache.hadoop.fs.s3a.Constants.SERVER_SIDE_ENCRYPTION_ALGORITHM;
+import static org.apache.hadoop.fs.s3a.Constants.SERVER_SIDE_ENCRYPTION_KEY;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.skipIfEncryptionTestsDisabled;
 
 /**
  * Test whether or not encryption works by turning it on. Some checks
@@ -42,11 +46,24 @@ public abstract class AbstractTestS3AEncryption extends AbstractS3ATestBase {
   protected Configuration createConfiguration() {
     Configuration conf = super.createConfiguration();
     S3ATestUtils.disableFilesystemCaching(conf);
-    conf.set(Constants.SERVER_SIDE_ENCRYPTION_ALGORITHM,
-            getSSEAlgorithm().getMethod());
+    patchConfigurationEncryptionSettings(conf);
     return conf;
   }
 
+  /**
+   * This removes the encryption settings from the
+   * configuration and then sets the
+   * fs.s3a.server-side-encryption-algorithm value to
+   * be that of {@code getSSEAlgorithm()}.
+   * Called in {@code createConfiguration()}.
+   * @param conf configuration to patch.
+   */
+  protected void patchConfigurationEncryptionSettings(
+      final Configuration conf) {
+    conf.set(SERVER_SIDE_ENCRYPTION_ALGORITHM,
+            getSSEAlgorithm().getMethod());
+  }
+
   private static final int[] SIZES = {
       0, 1, 2, 3, 4, 5, 254, 255, 256, 257, 2 ^ 12 - 1
   };
@@ -66,10 +83,15 @@ public void testEncryptionOverRename() throws Throwable {
     S3AFileSystem fs = getFileSystem();
     writeDataset(fs, src, data, data.length, 1024 * 1024, true);
     ContractTestUtils.verifyFileContents(fs, src, data);
-    Path dest = path(src.getName() + "-copy");
-    fs.rename(src, dest);
-    ContractTestUtils.verifyFileContents(fs, dest, data);
-    assertEncrypted(dest);
+    // this file will be encrypted
+    assertEncrypted(src);
+
+    Path targetDir = path("target");
+    mkdirs(targetDir);
+    fs.rename(src, targetDir);
+    Path renamedFile = new Path(targetDir, src.getName());
+    ContractTestUtils.verifyFileContents(fs, renamedFile, data);
+    assertEncrypted(renamedFile);
   }
 
   protected void validateEncryptionForFilesize(int len) throws IOException {
@@ -95,40 +117,14 @@ protected String createFilename(String name) {
    * @throws IOException on a failure
    */
   protected void assertEncrypted(Path path) throws IOException {
-    ObjectMetadata md = getFileSystem().getObjectMetadata(path);
-    switch(getSSEAlgorithm()) {
-    case SSE_C:
-      assertEquals("AES256", md.getSSECustomerAlgorithm());
-      String md5Key = convertKeyToMd5();
-      assertEquals(md5Key, md.getSSECustomerKeyMd5());
-      break;
-    case SSE_KMS:
-      assertEquals("aws:kms", md.getSSEAlgorithm());
-      //S3 will return full arn of the key, so specify global arn in properties
-      assertEquals(this.getConfiguration().
-          getTrimmed(Constants.SERVER_SIDE_ENCRYPTION_KEY),
-          md.getSSEAwsKmsKeyId());
-      break;
-    default:
-      assertEquals("AES256", md.getSSEAlgorithm());
-    }
-  }
-
-  /**
-   * Decodes the SERVER_SIDE_ENCRYPTION_KEY from base64 into an AES key, then
-   * gets the md5 of it, then encodes it in base64 so it will match the version
-   * that AWS returns to us.
-   *
-   * @return md5'd base64 encoded representation of the server side encryption
-   * key
-   */
-  private String convertKeyToMd5() {
-    String base64Key = getConfiguration().getTrimmed(
-        Constants.SERVER_SIDE_ENCRYPTION_KEY
-    );
-    byte[] key = Base64.decodeBase64(base64Key);
-    byte[] md5 =  DigestUtils.md5(key);
-    return Base64.encodeBase64String(md5).trim();
+    //S3 will return full arn of the key, so specify global arn in properties
+    String kmsKeyArn = this.getConfiguration().
+        getTrimmed(SERVER_SIDE_ENCRYPTION_KEY);
+    S3AEncryptionMethods algorithm = getSSEAlgorithm();
+    EncryptionTestUtils.assertEncrypted(getFileSystem(),
+            path,
+            algorithm,
+            kmsKeyArn);
   }
 
   protected abstract S3AEncryptionMethods getSSEAlgorithm();

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/EncryptionTestUtils.java

@@ -0,0 +1,102 @@
+/*
+ * 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;
+
+import java.io.IOException;
+
+import com.amazonaws.services.s3.model.ObjectMetadata;
+
+import org.apache.commons.codec.digest.DigestUtils;
+import org.apache.commons.net.util.Base64;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import static org.apache.hadoop.fs.s3a.Constants.SERVER_SIDE_ENCRYPTION_KEY;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+public final class EncryptionTestUtils {
+
+  /** Private constructor */
+  private EncryptionTestUtils() {
+  }
+
+  public static final String AWS_KMS_SSE_ALGORITHM = "aws:kms";
+
+  public static final String SSE_C_ALGORITHM = "AES256";
+
+  /**
+   * Decodes the SERVER_SIDE_ENCRYPTION_KEY from base64 into an AES key, then
+   * gets the md5 of it, then encodes it in base64 so it will match the version
+   * that AWS returns to us.
+   *
+   * @return md5'd base64 encoded representation of the server side encryption
+   * key
+   */
+  public static String convertKeyToMd5(FileSystem fs) {
+    String base64Key = fs.getConf().getTrimmed(
+            SERVER_SIDE_ENCRYPTION_KEY
+    );
+    byte[] key = Base64.decodeBase64(base64Key);
+    byte[] md5 =  DigestUtils.md5(key);
+    return Base64.encodeBase64String(md5).trim();
+  }
+
+  /**
+   * Assert that a path is encrypted with right encryption settings.
+   * @param path file path.
+   * @param algorithm encryption algorithm.
+   * @param kmsKeyArn full kms key.
+   */
+  public static void assertEncrypted(S3AFileSystem fs,
+                                     final Path path,
+                                     final S3AEncryptionMethods algorithm,
+                                     final String kmsKeyArn)
+          throws IOException {
+    ObjectMetadata md = fs.getObjectMetadata(path);
+    String details = String.format(
+            "file %s with encryption algorithm %s and key %s",
+            path,
+            md.getSSEAlgorithm(),
+            md.getSSEAwsKmsKeyId());
+    switch(algorithm) {
+    case SSE_C:
+      assertNull("Metadata algorithm should have been null in "
+                      + details,
+              md.getSSEAlgorithm());
+      assertEquals("Wrong SSE-C algorithm in "
+                      + details,
+              SSE_C_ALGORITHM, md.getSSECustomerAlgorithm());
+      String md5Key = convertKeyToMd5(fs);
+      assertEquals("getSSECustomerKeyMd5() wrong in " + details,
+              md5Key, md.getSSECustomerKeyMd5());
+      break;
+    case SSE_KMS:
+      assertEquals("Wrong algorithm in " + details,
+              AWS_KMS_SSE_ALGORITHM, md.getSSEAlgorithm());
+      assertEquals("Wrong KMS key in " + details,
+              kmsKeyArn,
+              md.getSSEAwsKmsKeyId());
+      break;
+    default:
+      assertEquals("AES256", md.getSSEAlgorithm());
+    }
+  }
+
+}

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AEncryptionSSEKMSDefaultKey.java

@@ -51,7 +51,8 @@ protected S3AEncryptionMethods getSSEAlgorithm() {
   @Override
   protected void assertEncrypted(Path path) throws IOException {
     ObjectMetadata md = getFileSystem().getObjectMetadata(path);
-    assertEquals("aws:kms", md.getSSEAlgorithm());
+    assertEquals("SSE Algorithm", EncryptionTestUtils.AWS_KMS_SSE_ALGORITHM,
+            md.getSSEAlgorithm());
     assertThat(md.getSSEAwsKmsKeyId(), containsString("arn:aws:kms:"));
   }
 }