AWS: Support S3 DSSE-KMS encryption (apache#8370)

Author: aajisaka

aws/src/integration/java/org/apache/iceberg/aws/s3/TestS3FileIOIntegration.java

@@ -291,6 +291,21 @@ public void testServerSideKmsEncryptionWithDefaultKey() throws Exception {
             aliasListEntry -> assertThat(aliasListEntry.aliasName()).isEqualTo("alias/aws/s3"));
   }
 
+  @Test
+  public void testDualLayerServerSideKmsEncryption() throws Exception {
+    S3FileIOProperties properties = new S3FileIOProperties();
+    properties.setSseType(S3FileIOProperties.DSSE_TYPE_KMS);
+    properties.setSseKey(kmsKeyArn);
+    S3FileIO s3FileIO = new S3FileIO(clientFactory::s3, properties);
+    write(s3FileIO);
+    validateRead(s3FileIO);
+    GetObjectResponse response =
+        s3.getObject(GetObjectRequest.builder().bucket(bucketName).key(objectKey).build())
+            .response();
+    assertThat(response.serverSideEncryption()).isEqualTo(ServerSideEncryption.AWS_KMS_DSSE);
+    assertThat(response.ssekmsKeyId()).isEqualTo(kmsKeyArn);
+  }
+
   @Test
   public void testServerSideCustomEncryption() throws Exception {
     // generate key

aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIOProperties.java

@@ -92,6 +92,14 @@ public class S3FileIOProperties implements Serializable {
    */
   public static final String SSE_TYPE_KMS = "kms";
 
+  /**
+   * S3 DSSE-KMS encryption.
+   *
+   * <p>For more details:
+   * https://docs.aws.amazon.com/AmazonS3/latest/userguide/UsingDSSEncryption.html
+   */
+  public static final String DSSE_TYPE_KMS = "dsse-kms";
+
   /**
    * S3 SSE-S3 encryption.
    *
@@ -109,9 +117,9 @@ public class S3FileIOProperties implements Serializable {
   public static final String SSE_TYPE_CUSTOM = "custom";
 
   /**
-   * If S3 encryption type is SSE-KMS, input is a KMS Key ID or ARN. In case this property is not
-   * set, default key "aws/s3" is used. If encryption type is SSE-C, input is a custom base-64
-   * AES256 symmetric key.
+   * If S3 encryption type is SSE-KMS or DSSE-KMS, input is a KMS Key ID or ARN. In case this
+   * property is not set, default key "aws/s3" is used. If encryption type is SSE-C, input is a
+   * custom base-64 AES256 symmetric key.
    */
   public static final String SSE_KEY = "s3.sse.key";
 

aws/src/main/java/org/apache/iceberg/aws/s3/S3RequestUtil.java

@@ -111,6 +111,11 @@ static void configureEncryption(
         kmsKeySetter.apply(s3FileIOProperties.sseKey());
         break;
 
+      case S3FileIOProperties.DSSE_TYPE_KMS:
+        encryptionSetter.apply(ServerSideEncryption.AWS_KMS_DSSE);
+        kmsKeySetter.apply(s3FileIOProperties.sseKey());
+        break;
+
       case S3FileIOProperties.SSE_TYPE_S3:
         encryptionSetter.apply(ServerSideEncryption.AES256);
         break;

aws/src/test/java/org/apache/iceberg/aws/s3/TestS3RequestUtil.java

@@ -88,6 +88,25 @@ public void testConfigureServerSideKmsEncryption() {
     Assertions.assertThat(customMd5).isNull();
   }
 
+  @Test
+  public void testConfigureDualLayerServerSideKmsEncryption() {
+    S3FileIOProperties s3FileIOProperties = new S3FileIOProperties();
+    s3FileIOProperties.setSseType(S3FileIOProperties.DSSE_TYPE_KMS);
+    s3FileIOProperties.setSseKey("key");
+    S3RequestUtil.configureEncryption(
+        s3FileIOProperties,
+        this::setServerSideEncryption,
+        this::setKmsKeyId,
+        this::setCustomAlgorithm,
+        this::setCustomKey,
+        this::setCustomMd5);
+    Assertions.assertThat(serverSideEncryption).isEqualTo(ServerSideEncryption.AWS_KMS_DSSE);
+    Assertions.assertThat(kmsKeyId).isEqualTo("key");
+    Assertions.assertThat(customAlgorithm).isNull();
+    Assertions.assertThat(customKey).isNull();
+    Assertions.assertThat(customMd5).isNull();
+  }
+
   @Test
   public void testConfigureEncryptionSkipNullSetters() {
     S3FileIOProperties s3FileIOProperties = new S3FileIOProperties();

docs/docs/aws.md

@@ -319,14 +319,15 @@ Here are the configurations that users can tune related to this feature:
 
 * [SSE-S3](https://docs.aws.amazon.com/AmazonS3/latest/dev/UsingServerSideEncryption.html): When you use Server-Side Encryption with Amazon S3-Managed Keys (SSE-S3), each object is encrypted with a unique key. As an additional safeguard, it encrypts the key itself with a master key that it regularly rotates. Amazon S3 server-side encryption uses one of the strongest block ciphers available, 256-bit Advanced Encryption Standard (AES-256), to encrypt your data.
 * [SSE-KMS](https://docs.aws.amazon.com/AmazonS3/latest/dev/UsingKMSEncryption.html): Server-Side Encryption with Customer Master Keys (CMKs) Stored in AWS Key Management Service (SSE-KMS) is similar to SSE-S3, but with some additional benefits and charges for using this service. There are separate permissions for the use of a CMK that provides added protection against unauthorized access of your objects in Amazon S3. SSE-KMS also provides you with an audit trail that shows when your CMK was used and by whom. Additionally, you can create and manage customer managed CMKs or use AWS managed CMKs that are unique to you, your service, and your Region.
+* [DSSE-KMS](https://docs.aws.amazon.com/AmazonS3/latest/userguide/UsingDSSEncryption.html): Dual-layer Server-Side Encryption with AWS Key Management Service keys (DSSE-KMS) is similar to SSE-KMS, but applies two layers of encryption to objects when they are uploaded to Amazon S3. DSSE-KMS can be used to fulfill compliance standards that require you to apply multilayer encryption to your data and have full control of your encryption keys.
 * [SSE-C](https://docs.aws.amazon.com/AmazonS3/latest/dev/ServerSideEncryptionCustomerKeys.html): With Server-Side Encryption with Customer-Provided Keys (SSE-C), you manage the encryption keys and Amazon S3 manages the encryption, as it writes to disks, and decryption when you access your objects.
 
 To enable server side encryption, use the following configuration properties:
 
 | Property                          | Default                                  | Description                                            |
 | --------------------------------- | ---------------------------------------- | ------------------------------------------------------ |
-| s3.sse.type                       | `none`                                   | `none`, `s3`, `kms` or `custom`                        |
-| s3.sse.key                        | `aws/s3` for `kms` type, null otherwise  | A KMS Key ID or ARN for `kms` type, or a custom base-64 AES256 symmetric key for `custom` type.  |
+| s3.sse.type                       | `none`                                   | `none`, `s3`, `kms`, `dsse-kms` or `custom`            |
+| s3.sse.key                        | `aws/s3` for `kms` and `dsse-kms` types, null otherwise | A KMS Key ID or ARN for `kms` and `dsse-kms` types, or a custom base-64 AES256 symmetric key for `custom` type.  |
 | s3.sse.md5                        | null                                     | If SSE type is `custom`, this value must be set as the base-64 MD5 digest of the symmetric key to ensure integrity. |
 
 ### S3 Access Control List