HADOOP-19013. Adding x-amz-server-side-encryption-aws-kms-key-id in t…

…he get file attributes for S3A.

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

@@ -55,6 +55,9 @@ public interface AWSHeaders {
   /** Header for optional server-side encryption algorithm. */
   String SERVER_SIDE_ENCRYPTION = "x-amz-server-side-encryption";
 
+  /** Header for optional server-side encryption algorithm. */
+  String SERVER_SIDE_ENCRYPTION_AWS_KMS_KEY_ID = "x-amz-server-side-encryption-aws-kms-key-id";
+
   /** Range header for the get object request. */
   String RANGE = "Range";
 

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

@@ -47,6 +47,7 @@
 import static org.apache.hadoop.fs.s3a.Statistic.INVOCATION_XATTR_GET_NAMED;
 import static org.apache.hadoop.fs.s3a.Statistic.INVOCATION_XATTR_GET_NAMED_MAP;
 import static org.apache.hadoop.fs.s3a.commit.CommitConstants.X_HEADER_MAGIC_MARKER;
+import static org.apache.hadoop.fs.s3a.impl.AWSHeaders.SERVER_SIDE_ENCRYPTION_AWS_KMS_KEY_ID;
 import static org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.trackDuration;
 
 /**
@@ -185,6 +186,9 @@ public class HeaderProcessing extends AbstractStoreOperation {
   public static final String XA_SERVER_SIDE_ENCRYPTION =
       XA_HEADER_PREFIX + AWSHeaders.SERVER_SIDE_ENCRYPTION;
 
+  public static final String XA_ENCRYPTION_KEY_ID =
+      XA_HEADER_PREFIX + SERVER_SIDE_ENCRYPTION_AWS_KMS_KEY_ID;
+
   /**
    * Storage Class XAttr: {@value}.
    */
@@ -363,6 +367,8 @@ private Map<String, byte[]> retrieveHeaders(
         md.versionId());
     maybeSetHeader(headers, XA_SERVER_SIDE_ENCRYPTION,
         md.serverSideEncryptionAsString());
+    maybeSetHeader(headers, XA_ENCRYPTION_KEY_ID,
+            md.ssekmsKeyId());
     maybeSetHeader(headers, XA_STORAGE_CLASS,
         md.storageClassAsString());
 

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

@@ -19,7 +19,13 @@
 package org.apache.hadoop.fs.s3a;
 
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+import java.util.Optional;
 
+import org.apache.hadoop.fs.s3a.impl.HeaderProcessing;
+import org.apache.hadoop.io.IOUtils;
+import org.assertj.core.api.Assertions;
 import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
 
 import org.apache.commons.codec.digest.DigestUtils;
@@ -28,6 +34,8 @@
 import org.apache.hadoop.fs.Path;
 
 import static org.apache.hadoop.fs.s3a.Constants.S3_ENCRYPTION_KEY;
+import static org.apache.hadoop.fs.s3a.impl.HeaderProcessing.XA_ENCRYPTION_KEY_ID;
+import static org.apache.hadoop.fs.s3a.impl.HeaderProcessing.XA_SERVER_SIDE_ENCRYPTION;
 import static org.assertj.core.api.Assertions.assertThat;
 
 public final class EncryptionTestUtils {
@@ -111,4 +119,27 @@ public static void assertEncrypted(S3AFileSystem fs,
     }
   }
 
+  /**
+   * Assert that a path is encrypted with right encryption settings.
+   * @param fs filesystem.
+   * @param path path
+   * @param algorithm encryption algorithm.
+   * @param kmsKey full kms key if present.
+   * @throws IOException any IOE.
+   */
+  public static void validateEncryptionFileAttributes(S3AFileSystem fs,
+                                                Path path,
+                                                String algorithm,
+                                                Optional<String> kmsKey) throws IOException {
+    Map<String, byte[]> xAttrs = fs.getXAttrs(path);
+    Assertions.assertThat(HeaderProcessing.decodeBytes(xAttrs.get(XA_SERVER_SIDE_ENCRYPTION)))
+                    .describedAs("Server side encryption algorithm must match")
+                    .isEqualTo(algorithm);
+    Assertions.assertThat(xAttrs.containsKey(XA_ENCRYPTION_KEY_ID))
+            .describedAs("Encryption key id should be present")
+            .isTrue();
+    kmsKey.ifPresent(s -> Assertions.assertThat(HeaderProcessing.decodeBytes(xAttrs.get(XA_ENCRYPTION_KEY_ID)))
+              .describedAs("Encryption key id should match with the kms key")
+              .isEqualTo(s));
+  }
 }

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

@@ -19,12 +19,20 @@
 package org.apache.hadoop.fs.s3a;
 
 import java.io.IOException;
+import java.util.Optional;
 
+import org.apache.hadoop.fs.contract.ContractTestUtils;
+import org.junit.Test;
 import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 
+import static org.apache.hadoop.fs.contract.ContractTestUtils.dataset;
+import static org.apache.hadoop.fs.contract.ContractTestUtils.writeDataset;
+import static org.apache.hadoop.fs.s3a.EncryptionTestUtils.validateEncryptionFileAttributes;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.getTestBucketName;
+import static org.apache.hadoop.fs.s3a.S3AUtils.getS3EncryptionKey;
 import static org.hamcrest.CoreMatchers.containsString;
 
 /**
@@ -56,4 +64,15 @@ protected void assertEncrypted(Path path) throws IOException {
             md.serverSideEncryptionAsString());
     assertThat(md.ssekmsKeyId(), containsString("arn:aws:kms:"));
   }
+
+  @Test
+  public void testEncryptionFileAttributes() throws Exception {
+    Path path = path(createFilename(1024));
+    byte[] data = dataset(1024, 'a', 'z');
+    S3AFileSystem fs = getFileSystem();
+    writeDataset(fs, path, data, data.length, 1024 * 1024, true);
+    ContractTestUtils.verifyFileContents(fs, path, data);
+    // we don't know the KMS key in case of server default option.
+    validateEncryptionFileAttributes(fs, path, EncryptionTestUtils.AWS_KMS_SSE_ALGORITHM,  Optional.empty());
+  }
 }

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

@@ -19,7 +19,11 @@
 package org.apache.hadoop.fs.s3a;
 
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+import java.util.Optional;
 
+import org.assertj.core.api.Assertions;
 import org.junit.Ignore;
 import org.junit.Test;
 
@@ -36,11 +40,14 @@
 import static org.apache.hadoop.fs.s3a.Constants.S3_ENCRYPTION_ALGORITHM;
 import static org.apache.hadoop.fs.s3a.Constants.SERVER_SIDE_ENCRYPTION_ALGORITHM;
 import static org.apache.hadoop.fs.s3a.EncryptionTestUtils.AWS_KMS_SSE_ALGORITHM;
+import static org.apache.hadoop.fs.s3a.EncryptionTestUtils.validateEncryptionFileAttributes;
 import static org.apache.hadoop.fs.s3a.S3AEncryptionMethods.SSE_KMS;
 import static org.apache.hadoop.fs.s3a.S3ATestUtils.getTestBucketName;
 import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
 import static org.apache.hadoop.fs.s3a.S3ATestUtils.skipIfEncryptionNotSet;
 import static org.apache.hadoop.fs.s3a.S3AUtils.getS3EncryptionKey;
+import static org.apache.hadoop.fs.s3a.impl.HeaderProcessing.XA_ENCRYPTION_KEY_ID;
+import static org.apache.hadoop.fs.s3a.impl.HeaderProcessing.XA_SERVER_SIDE_ENCRYPTION;
 
 /**
  * Concrete class that extends {@link AbstractTestS3AEncryption}
@@ -97,6 +104,21 @@ protected void assertEncrypted(Path path) throws IOException {
     EncryptionTestUtils.assertEncrypted(fs, path, SSE_KMS, kmsKey);
   }
 
+  @Test
+  public void testEncryptionFileAttributes() throws Exception {
+    Path path = path(createFilename(1024));
+    byte[] data = dataset(1024, 'a', 'z');
+    S3AFileSystem fs = getFileSystem();
+    writeDataset(fs, path, data, data.length, 1024 * 1024, true);
+    ContractTestUtils.verifyFileContents(fs, path, data);
+    Configuration c = fs.getConf();
+    String kmsKey = getS3EncryptionKey(getTestBucketName(c), c);
+    validateEncryptionFileAttributes(fs, path, AWS_KMS_SSE_ALGORITHM, Optional.of(kmsKey));
+  }
+
+
+
+
   @Override
   @Ignore
   @Test