confluentinc#770 - Add x-amz-server-side-encryption header for SSE/AE…

…S256 without customer key
85danf · Aug 8, 2024 · 5435ad8 · 5435ad8
1 parent eaa37cc
commit 5435ad8
Showing 1 changed file with 15 additions and 2 deletions.
diff --git a/kafka-connect-s3/src/main/java/io/confluent/connect/s3/storage/S3OutputStream.java b/kafka-connect-s3/src/main/java/io/confluent/connect/s3/storage/S3OutputStream.java
@@ -23,6 +23,7 @@
 import com.amazonaws.services.s3.model.CannedAccessControlList;
 import com.amazonaws.services.s3.model.CompleteMultipartUploadRequest;
 import com.amazonaws.services.s3.model.InitiateMultipartUploadRequest;
+import com.amazonaws.services.s3.model.ObjectMetadata;
 import com.amazonaws.services.s3.model.PartETag;
 import com.amazonaws.services.s3.model.SSEAlgorithm;
 import com.amazonaws.services.s3.model.SSEAwsKeyManagementParams;
@@ -205,14 +206,26 @@ private void internalClose() throws IOException {
     super.close();
   }
 
+  private ObjectMetadata newObjectMetadata() {
+    ObjectMetadata meta = new ObjectMetadata();
+    if (StringUtils.isNotBlank(ssea)) {
+      meta.setSSEAlgorithm(ssea);
+    }
+    return meta;
+  }
+
   private MultipartUpload newMultipartUpload() throws IOException {
     InitiateMultipartUploadRequest initRequest = new InitiateMultipartUploadRequest(
         bucket,
         key
     ).withCannedACL(cannedAcl);
 
-    if (SSEAlgorithm.KMS.toString().equalsIgnoreCase(ssea)
-        && StringUtils.isNotBlank(sseKmsKeyId)) {
+    if (SSEAlgorithm.AES256.toString().equalsIgnoreCase(ssea)
+        && sseCustomerKey == null) {
+      log.debug("Using SSE (AES256) without customer key");
+      initRequest.setObjectMetadata(newObjectMetadata());
+    } else if (SSEAlgorithm.KMS.toString().equalsIgnoreCase(ssea)
+               && StringUtils.isNotBlank(sseKmsKeyId)) {
       log.debug("Using KMS Key ID: {}", sseKmsKeyId);
       initRequest.setSSEAwsKeyManagementParams(new SSEAwsKeyManagementParams(sseKmsKeyId));
     } else if (sseCustomerKey != null) {