Revert "feat: Add S3CrtAsyncClient as MultipartPutobject (#90)"

This reverts commit 24be141.

Author: J Plasmeier

pom.xml

@@ -73,13 +73,6 @@
       <optional>true</optional>
     </dependency>
 
-    <dependency>
-      <groupId>software.amazon.awssdk.crt</groupId>
-      <artifactId>aws-crt</artifactId>
-      <version>0.21.5</version>
-      <optional>true</optional>
-    </dependency>
-
     <dependency>
       <groupId>net.jcip</groupId>
       <artifactId>jcip-annotations</artifactId>

src/main/java/software/amazon/encryption/s3/S3AsyncEncryptionClient.java

@@ -40,22 +40,18 @@
 public class S3AsyncEncryptionClient extends DelegatingS3AsyncClient {
 
     private final S3AsyncClient _wrappedClient;
-    private final S3AsyncClient _wrappedCrtClient;
     private final CryptographicMaterialsManager _cryptoMaterialsManager;
     private final SecureRandom _secureRandom;
     private final boolean _enableLegacyWrappingAlgorithms;
     private final boolean _enableLegacyUnauthenticatedModes;
     private final boolean _enableDelayedAuthenticationMode;
-    private final boolean _enableMultipartPutObject;
 
     private S3AsyncEncryptionClient(Builder builder) {
         super(builder._wrappedClient);
         _wrappedClient = builder._wrappedClient;
-        _wrappedCrtClient = builder._wrappedCrtClient;
         _cryptoMaterialsManager = builder._cryptoMaterialsManager;
         _secureRandom = builder._secureRandom;
         _enableLegacyWrappingAlgorithms = builder._enableLegacyWrappingAlgorithms;
-        _enableMultipartPutObject = builder._enableMultipartPutObject;
         _enableLegacyUnauthenticatedModes = builder._enableLegacyUnauthenticatedModes;
         _enableDelayedAuthenticationMode = builder._enableDelayedAuthenticationMode;
     }
@@ -75,8 +71,6 @@ public CompletableFuture<PutObjectResponse> putObject(PutObjectRequest putObject
             throws AwsServiceException, SdkClientException {
         PutEncryptedObjectPipeline pipeline = PutEncryptedObjectPipeline.builder()
                 .s3AsyncClient(_wrappedClient)
-                .crtClient(_wrappedCrtClient)
-                .enableMultipartPutObject(_enableMultipartPutObject)
                 .cryptoMaterialsManager(_cryptoMaterialsManager)
                 .secureRandom(_secureRandom)
                 .build();
@@ -133,16 +127,15 @@ public void close() {
     // TODO: The async / non-async clients can probably share a builder - revisit after implementing async
     public static class Builder {
         private S3AsyncClient _wrappedClient = S3AsyncClient.builder().build();
-        private S3AsyncClient _wrappedCrtClient = null;
         private CryptographicMaterialsManager _cryptoMaterialsManager;
         private Keyring _keyring;
         private SecretKey _aesKey;
         private PartialRsaKeyPair _rsaKeyPair;
         private String _kmsKeyId;
         private boolean _enableLegacyWrappingAlgorithms = false;
         private boolean _enableLegacyUnauthenticatedModes = false;
-        private boolean _enableMultipartPutObject = false;
         private boolean _enableDelayedAuthenticationMode = false;
+        private boolean _enableMultipartPutObject = false;
         private Provider _cryptoProvider = null;
         private SecureRandom _secureRandom = new SecureRandom();
 
@@ -158,8 +151,7 @@ public Builder wrappedClient(S3AsyncClient wrappedClient) {
             if (wrappedClient instanceof S3AsyncEncryptionClient) {
                 throw new S3EncryptionClientException("Cannot use S3EncryptionClient as wrapped client");
             }
-            // Initializes only when wrappedAsyncClient is configured by user.
-            this._wrappedCrtClient = wrappedClient;
+
             this._wrappedClient = wrappedClient;
             return this;
         }
@@ -246,7 +238,9 @@ public Builder enableDelayedAuthenticationMode(boolean shouldEnableDelayedAuthen
         }
 
         public Builder enableMultipartPutObject(boolean _enableMultipartPutObject) {
-            this._enableMultipartPutObject = _enableMultipartPutObject;
+            if (_enableMultipartPutObject) {
+                throw new S3EncryptionClientException("Async multipart PutObject is currently disabled.");
+            }
             return this;
         }
 

src/main/java/software/amazon/encryption/s3/S3EncryptionClient.java

@@ -18,6 +18,7 @@
 import software.amazon.awssdk.services.s3.model.AbortMultipartUploadResponse;
 import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest;
 import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadResponse;
+import software.amazon.awssdk.services.s3.model.CompletedPart;
 import software.amazon.awssdk.services.s3.model.CreateMultipartUploadRequest;
 import software.amazon.awssdk.services.s3.model.CreateMultipartUploadResponse;
 import software.amazon.awssdk.services.s3.model.DeleteObjectRequest;
@@ -32,25 +33,33 @@
 import software.amazon.awssdk.services.s3.model.UploadPartRequest;
 import software.amazon.awssdk.services.s3.model.UploadPartResponse;
 import software.amazon.encryption.s3.internal.GetEncryptedObjectPipeline;
+import software.amazon.encryption.s3.internal.MultiFileOutputStream;
 import software.amazon.encryption.s3.internal.MultipartUploadObjectPipeline;
 import software.amazon.encryption.s3.internal.PutEncryptedObjectPipeline;
+import software.amazon.encryption.s3.internal.UploadObjectObserver;
 import software.amazon.encryption.s3.materials.AesKeyring;
 import software.amazon.encryption.s3.materials.CryptographicMaterialsManager;
 import software.amazon.encryption.s3.materials.DefaultCryptoMaterialsManager;
 import software.amazon.encryption.s3.materials.Keyring;
 import software.amazon.encryption.s3.materials.KmsKeyring;
+import software.amazon.encryption.s3.materials.MultipartConfiguration;
 import software.amazon.encryption.s3.materials.PartialRsaKeyPair;
 import software.amazon.encryption.s3.materials.RsaKeyring;
 
 import javax.crypto.SecretKey;
+import java.io.IOException;
 import java.security.KeyPair;
 import java.security.Provider;
 import java.security.SecureRandom;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.function.Consumer;
 
 import static software.amazon.encryption.s3.S3EncryptionClientUtilities.INSTRUCTION_FILE_SUFFIX;
@@ -64,12 +73,13 @@ public class S3EncryptionClient extends DelegatingS3Client {
 
     // Used for request-scoped encryption contexts for supporting keys
     public static final ExecutionAttribute<Map<String, String>> ENCRYPTION_CONTEXT = new ExecutionAttribute<>("EncryptionContext");
+    public static final ExecutionAttribute<MultipartConfiguration> CONFIGURATION = new ExecutionAttribute<>("MultipartConfiguration");
     // TODO: Replace with UploadPartRequest.isLastPart() when launched.
     // Used for multipart uploads
     public static final ExecutionAttribute<Boolean> IS_LAST_PART = new ExecutionAttribute<>("isLastPart");
 
-    private final S3AsyncClient _wrappedClient;
-    private final S3AsyncClient _wrappedCrtClient;
+    private final S3Client _wrappedClient;
+    private final S3AsyncClient _wrappedAsyncClient;
     private final CryptographicMaterialsManager _cryptoMaterialsManager;
     private final SecureRandom _secureRandom;
     private final boolean _enableLegacyWrappingAlgorithms;
@@ -79,11 +89,9 @@ public class S3EncryptionClient extends DelegatingS3Client {
     private final MultipartUploadObjectPipeline _multipartPipeline;
 
     private S3EncryptionClient(Builder builder) {
-        // The non-encrypted APIs will use a default client.
-        // In the future, we may want to make this configurable.
-        super(S3Client.create());
+        super(builder._wrappedClient);
         _wrappedClient = builder._wrappedClient;
-        _wrappedCrtClient = builder._wrappedCrtClient;
+        _wrappedAsyncClient = builder._wrappedAsyncClient;
         _cryptoMaterialsManager = builder._cryptoMaterialsManager;
         _secureRandom = builder._secureRandom;
         _enableLegacyWrappingAlgorithms = builder._enableLegacyWrappingAlgorithms;
@@ -103,6 +111,13 @@ public static Consumer<AwsRequestOverrideConfiguration.Builder> withAdditionalCo
                 builder.putExecutionAttribute(S3EncryptionClient.ENCRYPTION_CONTEXT, encryptionContext);
     }
 
+    // Helper function to attach encryption contexts to a request
+    public static Consumer<AwsRequestOverrideConfiguration.Builder> withAdditionalConfiguration(Map<String, String> encryptionContext, MultipartConfiguration multipartConfiguration) {
+        return builder ->
+                builder.putExecutionAttribute(S3EncryptionClient.ENCRYPTION_CONTEXT, encryptionContext)
+                        .putExecutionAttribute(S3EncryptionClient.CONFIGURATION, multipartConfiguration);
+    }
+
     // Helper function to determine last upload part during multipart uploads
     public static Consumer<AwsRequestOverrideConfiguration.Builder> isLastPart(Boolean isLastPart) {
         return builder ->
@@ -113,11 +128,21 @@ public static Consumer<AwsRequestOverrideConfiguration.Builder> isLastPart(Boole
     public PutObjectResponse putObject(PutObjectRequest putObjectRequest, RequestBody requestBody)
             throws AwsServiceException, SdkClientException {
 
+        if (_enableMultipartPutObject) {
+            try {
+                // TODO: Confirm best way to wrap CompleteMultipartUploadResponse with PutObjectResponse
+                CompleteMultipartUploadResponse completeResponse = multipartPutObject(putObjectRequest, requestBody);
+                PutObjectResponse response = PutObjectResponse.builder()
+                        .eTag(completeResponse.eTag())
+                        .build();
+                return response;
+            } catch (Throwable e) {
+                throw new S3EncryptionClientException("Exception while performing Multipart Upload PutObject", e);
+            }
+        }
         PutEncryptedObjectPipeline pipeline = PutEncryptedObjectPipeline.builder()
-                .s3AsyncClient(_wrappedClient)
-                .crtClient(_wrappedCrtClient)
+                .s3AsyncClient(_wrappedAsyncClient)
                 .cryptoMaterialsManager(_cryptoMaterialsManager)
-                .enableMultipartPutObject(_enableMultipartPutObject)
                 .secureRandom(_secureRandom)
                 .build();
 
@@ -131,7 +156,7 @@ public <T> T getObject(GetObjectRequest getObjectRequest,
             throws AwsServiceException, SdkClientException {
 
         GetEncryptedObjectPipeline pipeline = GetEncryptedObjectPipeline.builder()
-                .s3AsyncClient(_wrappedClient)
+                .s3AsyncClient(_wrappedAsyncClient)
                 .cryptoMaterialsManager(_cryptoMaterialsManager)
                 .enableLegacyWrappingAlgorithms(_enableLegacyWrappingAlgorithms)
                 .enableLegacyUnauthenticatedModes(_enableLegacyUnauthenticatedModes)
@@ -148,14 +173,76 @@ public <T> T getObject(GetObjectRequest getObjectRequest,
         }
     }
 
+    private CompleteMultipartUploadResponse multipartPutObject(PutObjectRequest request, RequestBody requestBody) throws Throwable {
+
+        AwsRequestOverrideConfiguration overrideConfig = request.overrideConfiguration().get();
+        // If MultipartConfiguration is null, Initialize MultipartConfiguration
+        MultipartConfiguration multipartConfiguration = overrideConfig
+                .executionAttributes()
+                .getOptionalAttribute(S3EncryptionClient.CONFIGURATION)
+                .orElse(MultipartConfiguration.builder().build());
+
+        ExecutorService es = multipartConfiguration.executorService();
+        final boolean defaultExecutorService = es == null;
+        if (es == null) {
+            throw new S3EncryptionClientException("ExecutorService should not be null, Please initialize during MultipartConfiguration");
+        }
+
+        UploadObjectObserver observer = multipartConfiguration.uploadObjectObserver();
+        if (observer == null) {
+            throw new S3EncryptionClientException("UploadObjectObserver should not be null, Please initialize during MultipartConfiguration");
+        }
+
+        observer.init(request, _wrappedAsyncClient, this, es);
+        final String uploadId = observer.onUploadCreation(request);
+        final List<CompletedPart> partETags = new ArrayList<>();
+
+        MultiFileOutputStream outputStream = multipartConfiguration.multiFileOutputStream();
+        if (outputStream == null) {
+            throw new S3EncryptionClientException("MultiFileOutputStream should not be null, Please initialize during MultipartConfiguration");
+        }
+
+        try {
+            // initialize the multi-file output stream
+            outputStream.init(observer, multipartConfiguration.partSize(), multipartConfiguration.diskLimit());
+            // Kicks off the encryption-upload pipeline;
+            // Note outputStream is automatically closed upon method completion.
+            _multipartPipeline.putLocalObject(requestBody, uploadId, outputStream);
+            // block till all part have been uploaded
+            for (Future<Map<Integer, UploadPartResponse>> future : observer.futures()) {
+                Map<Integer, UploadPartResponse> partResponseMap = future.get();
+                partResponseMap.forEach((partNumber, uploadPartResponse) -> partETags.add(CompletedPart.builder()
+                        .partNumber(partNumber)
+                        .eTag(uploadPartResponse.eTag())
+                        .build()));
+            }
+        } catch (IOException | InterruptedException | ExecutionException | RuntimeException | Error ex) {
+            throw onAbort(observer, ex);
+        } finally {
+            if (defaultExecutorService) {
+                // shut down the locally created thread pool
+                es.shutdownNow();
+            }
+            // delete left-over temp files
+            outputStream.cleanup();
+        }
+        // Complete upload
+        return observer.onCompletion(partETags);
+    }
+
+    private <T extends Throwable> T onAbort(UploadObjectObserver observer, T t) {
+        observer.onAbort();
+        return t;
+    }
+
     @Override
     public DeleteObjectResponse deleteObject(DeleteObjectRequest deleteObjectRequest) throws AwsServiceException,
             SdkClientException {
         // Delete the object
-        DeleteObjectResponse deleteObjectResponse = _wrappedClient.deleteObject(deleteObjectRequest).join();
+        DeleteObjectResponse deleteObjectResponse = _wrappedAsyncClient.deleteObject(deleteObjectRequest).join();
         // If Instruction file exists, delete the instruction file as well.
         String instructionObjectKey = deleteObjectRequest.key() + INSTRUCTION_FILE_SUFFIX;
-        _wrappedClient.deleteObject(builder -> builder
+        _wrappedAsyncClient.deleteObject(builder -> builder
                 .bucket(deleteObjectRequest.bucket())
                 .key(instructionObjectKey)).join();
         return deleteObjectResponse;
@@ -165,10 +252,10 @@ public DeleteObjectResponse deleteObject(DeleteObjectRequest deleteObjectRequest
     public DeleteObjectsResponse deleteObjects(DeleteObjectsRequest deleteObjectsRequest) throws AwsServiceException,
             SdkClientException {
         // Delete the objects
-        DeleteObjectsResponse deleteObjectsResponse = _wrappedClient.deleteObjects(deleteObjectsRequest).join();
+        DeleteObjectsResponse deleteObjectsResponse = _wrappedAsyncClient.deleteObjects(deleteObjectsRequest).join();
         // If Instruction files exists, delete the instruction files as well.
         List<ObjectIdentifier> deleteObjects = instructionFileKeysToDelete(deleteObjectsRequest);
-        _wrappedClient.deleteObjects(DeleteObjectsRequest.builder()
+        _wrappedAsyncClient.deleteObjects(DeleteObjectsRequest.builder()
                 .bucket(deleteObjectsRequest.bucket())
                 .delete(builder -> builder.objects(deleteObjects))
                 .build()).join();
@@ -213,12 +300,14 @@ public AbortMultipartUploadResponse abortMultipartUpload(AbortMultipartUploadReq
 
     @Override
     public void close() {
-        _wrappedClient.close();
+        _wrappedAsyncClient.close();
     }
 
     public static class Builder {
-        private S3AsyncClient _wrappedClient = S3AsyncClient.create();
-        private S3AsyncClient _wrappedCrtClient = null;
+        // The non-encrypted APIs will use a default client.
+        // In the future, we may want to make this configurable.
+        private S3Client _wrappedClient = S3Client.create();
+        private S3AsyncClient _wrappedAsyncClient = S3AsyncClient.create();
 
         private MultipartUploadObjectPipeline _multipartPipeline;
         private CryptographicMaterialsManager _cryptoMaterialsManager;
@@ -241,13 +330,22 @@ private Builder() {
          * S3AsyncClient will be reflected in this Builder.
          */
         @SuppressFBWarnings(value = "EI_EXPOSE_REP2", justification = "Pass mutability into wrapping client")
-        public Builder wrappedClient(S3AsyncClient wrappedClient) {
-            if (wrappedClient instanceof S3AsyncEncryptionClient) {
-                throw new S3EncryptionClientException("Cannot use S3AsyncEncryptionClient as wrapped client");
+        public Builder wrappedAsyncClient(S3AsyncClient _wrappedAsyncClient) {
+            if (_wrappedAsyncClient instanceof S3AsyncEncryptionClient) {
+                throw new S3EncryptionClientException("Cannot use S3EncryptionClient as wrapped client");
+            }
+
+            this._wrappedAsyncClient = _wrappedAsyncClient;
+            return this;
+        }
+
+        @SuppressFBWarnings(value = "EI_EXPOSE_REP2", justification = "Pass mutability into wrapping client")
+        public Builder _wrappedAsyncClient(S3AsyncClient _wrappedAsyncClient) {
+            if (_wrappedAsyncClient instanceof S3AsyncEncryptionClient) {
+                throw new S3EncryptionClientException("Cannot use S3EncryptionClient as wrapped client");
             }
-            // Initializes only when wrappedClient is configured by user.
-            this._wrappedCrtClient = wrappedClient;
-            this._wrappedClient = wrappedClient;
+
+            this._wrappedAsyncClient = _wrappedAsyncClient;
             return this;
         }
 
@@ -385,7 +483,7 @@ public S3EncryptionClient build() {
             }
 
             _multipartPipeline = MultipartUploadObjectPipeline.builder()
-                    .s3AsyncClient(_wrappedClient)
+                    .s3AsyncClient(_wrappedAsyncClient)
                     .cryptoMaterialsManager(_cryptoMaterialsManager)
                     .secureRandom(_secureRandom)
                     .build();

src/main/java/software/amazon/encryption/s3/internal/FileDeletionEvent.java

@@ -0,0 +1,9 @@
+package software.amazon.encryption.s3.internal;
+
+/**
+ * A file deletion event.
+ */
+public class FileDeletionEvent {
+    // currently only a placeholder so method signature won't be affected by
+    // future changes
+}