Merge pull request #30 from trocco-io/feat/aws-credential-utils

dmikurube · web-flow · commit 79d267c7a8e2 · 2022-11-28T15:05:53.000+09:00
Sync config with s3 input  for aws authmethod and http_proxy
diff --git a/README.md b/README.md
@@ -25,8 +25,62 @@
 - **tmp_path**: temporary file directory. If null, it is associated with the default FileSystem. (string, default: null)
 - **tmp_path_prefix**: prefix of temporary files (string, default: 'embulk-output-s3-')
 - **canned_acl**: canned access control list for created objects ([enum](#cannedaccesscontrollist), default: null)
-- **proxy_host**: proxy host to use when accessing AWS S3 via proxy. (string, default: null )
-- **proxy_port**: proxy port to use when accessing AWS S3 via proxy. (string, default: null )
+- [Deprecated] **proxy_host**: proxy host to use when accessing AWS S3 via proxy. (string, default: null )
+- [Deprecated] **proxy_port**: proxy port to use when accessing AWS S3 via proxy. (string, default: null )
+- **http_proxy**: http proxy configuration to use when accessing AWS S3 via http proxy. (optional)
+  - **host**: proxy host (string, required)
+  - **port**: proxy port (int, optional)
+  - **https**: use https or not (boolean, default true)
+  - **user**: proxy user (string, optional)
+  - **password**: proxy password (string, optional)
+
+- **auth_method**: name of mechanism to authenticate requests (basic, env, instance, profile, properties, anonymous, or session. default: basic)
+
+    - "basic": uses access_key_id and secret_access_key to authenticate.
+
+        - **access_key_id**: AWS access key ID (string, required)
+
+        - **secret_access_key**: AWS secret access key (string, required)
+
+    - "env": uses AWS_ACCESS_KEY_ID (or AWS_ACCESS_KEY) and AWS_SECRET_KEY (or AWS_SECRET_ACCESS_KEY) environment variables.
+
+    - "instance": uses EC2 instance profile.
+
+    - "profile": uses credentials written in a file. Format of the file is as following, where `[...]` is a name of profile.
+
+        - **profile_file**: path to a profiles file. (string, default: given by AWS_CREDENTIAL_PROFILES_FILE environment varialbe, or ~/.aws/credentials).
+
+        - **profile_name**: name of a profile. (string, default: `"default"`)
+
+      ```
+      [default]
+      aws_access_key_id=YOUR_ACCESS_KEY_ID
+      aws_secret_access_key=YOUR_SECRET_ACCESS_KEY
+  
+      [profile2]
+      ...
+      ```
+
+    - "properties": uses aws.accessKeyId and aws.secretKey Java system properties.
+
+    - "anonymous": uses anonymous access. This auth method can access only public files.
+
+    - "session": uses temporary-generated access_key_id, secret_access_key and session_token.
+
+        - **access_key_id**: AWS access key ID (string, required)
+
+        - **secret_access_key**: AWS secret access key (string, required)
+
+        - **session_token**: session token (string, required)
+
+    - "default": uses AWS SDK's default strategy to look up available credentials from runtime environment. This method behaves like the combination of the following methods.
+
+        1. "env"
+        1. "properties"
+        1. "profile"
+        1. "instance"
+
+
 
 ### CannedAccessControlList
 you can choose one of the below list.
diff --git a/build.gradle b/build.gradle
@@ -46,6 +46,29 @@ dependencies {
         exclude group: "commons-logging", module: "commons-logging"
     }
 
+    compile("com.amazonaws:aws-java-sdk-sts:1.11.466") {
+        // They conflict with embulk-core. They are once excluded here,
+        // and added explicitly with versions exactly the same with embulk-core:0.10.29.
+        exclude group: "com.fasterxml.jackson.core", module: "jackson-annotations"
+        exclude group: "com.fasterxml.jackson.core", module: "jackson-core"
+        exclude group: "com.fasterxml.jackson.core", module: "jackson-databind"
+        exclude group: "joda-time", module: "joda-time"
+
+        // commons-logging api is provided by jcl-over-slf4j below.
+        exclude group: "commons-logging", module: "commons-logging"
+    }
+
+    compile("org.embulk:embulk-util-aws-credentials:0.4.1") {
+        // They conflict with embulk-core. They are once excluded here,
+        // and added explicitly with versions exactly the same with embulk-core:0.10.29.
+        exclude group: "com.fasterxml.jackson.core", module: "jackson-annotations"
+        exclude group: "com.fasterxml.jackson.core", module: "jackson-core"
+        exclude group: "com.fasterxml.jackson.core", module: "jackson-databind"
+        exclude group: "com.fasterxml.jackson.datatype", module: "jackson-datatype-jdk8"
+        exclude group: "javax.validation", module: "validation-api"
+        exclude group: "joda-time", module: "joda-time"
+    }
+
     compile("org.embulk:embulk-util-config:0.3.1") {
         // Jackson libraries conflict with embulk-core before v0.10.32.
         // They are once excluded here, and re-added explicitly below.
diff --git a/gradle/dependency-locks/embulkPluginRuntime.lockfile b/gradle/dependency-locks/embulkPluginRuntime.lockfile
@@ -4,6 +4,7 @@
 com.amazonaws:aws-java-sdk-core:1.11.1034
 com.amazonaws:aws-java-sdk-kms:1.11.1034
 com.amazonaws:aws-java-sdk-s3:1.11.1034
+com.amazonaws:aws-java-sdk-sts:1.11.466
 com.amazonaws:jmespath-java:1.11.1034
 com.fasterxml.jackson.core:jackson-annotations:2.6.7
 com.fasterxml.jackson.core:jackson-core:2.6.7
@@ -18,6 +19,7 @@ javax.xml.bind:jaxb-api:2.2.11
 joda-time:joda-time:2.9.2
 org.apache.httpcomponents:httpclient:4.5.13
 org.apache.httpcomponents:httpcore:4.4.13
+org.embulk:embulk-util-aws-credentials:0.4.1
 org.embulk:embulk-util-config:0.3.1
 org.embulk:embulk-util-file:0.1.3
 org.slf4j:jcl-over-slf4j:1.7.12
diff --git a/src/main/java/org/embulk/output/s3/HttpProxy.java b/src/main/java/org/embulk/output/s3/HttpProxy.java
@@ -0,0 +1,55 @@
+/*
+ * Copyright 2022 The Embulk project
+ *
+ * Licensed 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.embulk.output.s3;
+
+
+import org.embulk.util.config.Config;
+import org.embulk.util.config.ConfigDefault;
+import org.embulk.util.config.Task;
+
+import java.util.Optional;
+
+/**
+ * HttpProxy is config unit for Input/Output plugins' configs.
+ *
+ * TODO: This unit will be moved to embulk/embulk-plugin-units.git.
+ * TODO: Consider using @JsonProperty(defaultValue=...) in Jackson 2.6+.
+ */
+public interface HttpProxy
+        extends Task
+{
+    @Config("host")
+    String getHost();
+    void setHost(String host);
+
+    @Config("port")
+    @ConfigDefault("null")
+    Optional<Integer> getPort();
+    void setPort(Optional<Integer> host);
+
+    @Config("https")
+    @ConfigDefault("true")
+    boolean getHttps();
+
+    @Config("user")
+    @ConfigDefault("null")
+    Optional<String> getUser();
+
+    @Config("password")
+    @ConfigDefault("null")
+    Optional<String> getPassword();
+}
diff --git a/src/main/java/org/embulk/output/s3/S3FileOutputPlugin.java b/src/main/java/org/embulk/output/s3/S3FileOutputPlugin.java
@@ -25,6 +25,12 @@
 import java.util.List;
 import java.util.Locale;
 
+import com.amazonaws.Protocol;
+import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.client.builder.AwsClientBuilder;
+import com.amazonaws.retry.PredefinedRetryPolicies;
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.AmazonS3ClientBuilder;
 import org.embulk.util.config.Config;
 import org.embulk.util.config.ConfigDefault;
 import org.embulk.config.ConfigDiff;
@@ -44,10 +50,10 @@
 import org.slf4j.Logger;
 
 import com.amazonaws.ClientConfiguration;
-import com.amazonaws.auth.BasicAWSCredentials;
-import com.amazonaws.services.s3.AmazonS3Client;
 import com.amazonaws.services.s3.model.CannedAccessControlList;
 import com.amazonaws.services.s3.model.PutObjectRequest;
+import org.embulk.util.aws.credentials.AwsCredentials;
+import org.embulk.util.aws.credentials.AwsCredentialsTask;
 import org.slf4j.LoggerFactory;
 
 import java.util.Optional;
@@ -63,7 +69,7 @@ public class S3FileOutputPlugin
     private static final ConfigMapper CONFIG_MAPPER = CONFIG_MAPPER_FACTORY.createConfigMapper();
 
     public interface PluginTask
-            extends Task
+            extends AwsCredentialsTask, Task
     {
         @Config("path_prefix")
         String getPathPrefix();
@@ -82,13 +88,10 @@ public interface PluginTask
         @ConfigDefault("null")
         Optional<String> getEndpoint();
 
-        @Config("access_key_id")
+        @Config("http_proxy")
         @ConfigDefault("null")
-        Optional<String> getAccessKeyId();
-
-        @Config("secret_access_key")
-        @ConfigDefault("null")
-        Optional<String> getSecretAccessKey();
+        Optional<HttpProxy> getHttpProxy();
+        void setHttpProxy(Optional<HttpProxy> httpProxy);
 
         @Config("proxy_host")
         @ConfigDefault("null")
@@ -109,6 +112,10 @@ public interface PluginTask
         @Config("canned_acl")
         @ConfigDefault("null")
         Optional<CannedAccessControlList> getCannedAccessControlList();
+
+        @Config("region")
+        @ConfigDefault("null")
+        Optional<String> getRegion();
     }
 
     public static class S3FileOutput
@@ -124,42 +131,122 @@ public static class S3FileOutput
 
         private int taskIndex;
         private int fileIndex;
-        private AmazonS3Client client;
+        private AmazonS3 client;
         private OutputStream current;
         private Path tempFilePath;
         private String tempPath = null;
 
-        private static AmazonS3Client newS3Client(PluginTask task)
+        private AmazonS3 newS3Client(final PluginTask task)
+        {
+            Optional<String> endpoint = task.getEndpoint();
+            Optional<String> region = task.getRegion();
+
+            final AmazonS3ClientBuilder builder = AmazonS3ClientBuilder
+                    .standard()
+                    .withCredentials(getCredentialsProvider(task))
+                    .withClientConfiguration(getClientConfiguration(task));
+
+            // Favor the `endpoint` configuration, then `region`, if both are absent then `s3.amazonaws.com` will be used.
+            if (endpoint.isPresent()) {
+                if (region.isPresent()) {
+                    logger.warn("Either configure endpoint or region, " +
+                            "if both is specified only the endpoint will be in effect.");
+                }
+                builder.setEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(endpoint.get(), null));
+            }
+            else if (region.isPresent()) {
+                builder.setRegion(region.get());
+            }
+            else {
+                // This is to keep the AWS SDK upgrading to 1.11.x to be backward compatible with old configuration.
+                //
+                // On SDK 1.10.x, when neither endpoint nor region is set explicitly, the client's endpoint will be by
+                // default `s3.amazonaws.com`. And for pre-Signature-V4, this will work fine as the bucket's region
+                // will be resolved to the appropriate region on server (AWS) side.
+                //
+                // On SDK 1.11.x, a region will be computed on client side by AwsRegionProvider and the endpoint now will
+                // be region-specific `<region>.s3.amazonaws.com` and might be the wrong one.
+                //
+                // So a default endpoint of `s3.amazonaws.com` when both endpoint and region configs are absent are
+                // necessary to make old configurations won't suddenly break. The side effect is that this will render
+                // AwsRegionProvider useless. And it's worth to note that Signature-V4 won't work with either versions with
+                // no explicit region or endpoint as the region (inferrable from endpoint) are necessary for signing.
+                builder.setEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration("s3.amazonaws.com", null));
+            }
+
+            builder.withForceGlobalBucketAccessEnabled(true);
+            return builder.build();
+        }
+
+        private AWSCredentialsProvider getCredentialsProvider(PluginTask task)
         {
-            AmazonS3Client client;
+            return AwsCredentials.getAWSCredentialsProvider(task);
+        }
+
+        private ClientConfiguration getClientConfiguration(PluginTask task)
+        {
+            ClientConfiguration clientConfig = new ClientConfiguration();
 
-            // TODO: Support more configurations.
-            ClientConfiguration config = new ClientConfiguration();
+            clientConfig.setMaxConnections(50); // SDK default: 50
+            clientConfig.setSocketTimeout(8 * 60 * 1000); // SDK default: 50*1000
+            clientConfig.setRetryPolicy(PredefinedRetryPolicies.NO_RETRY_POLICY);
 
+            // set http proxy
+            // backward compatibility
             if (task.getProxyHost().isPresent()) {
-                config.setProxyHost(task.getProxyHost().get());
+                logger.warn("Configuration with \"proxy_host\" is deprecated. Use \"http_proxy.host\" instead.");
+                if (!task.getHttpProxy().isPresent()) {
+                    ConfigMapper configMapper = CONFIG_MAPPER_FACTORY.createConfigMapper();
+                    ConfigSource configSource = CONFIG_MAPPER_FACTORY.newConfigSource();
+                    configSource.set("host", task.getProxyHost().get());
+                    HttpProxy httpProxy = configMapper.map(configSource, HttpProxy.class);
+                    task.setHttpProxy(Optional.of(httpProxy));
+                } else {
+                    HttpProxy httpProxy = task.getHttpProxy().get();
+                    if (httpProxy.getHost().isEmpty()) {
+                        httpProxy.setHost(task.getProxyHost().get());
+                        task.setHttpProxy(Optional.of(httpProxy));
+                    }
+                }
             }
 
             if (task.getProxyPort().isPresent()) {
-                config.setProxyPort(task.getProxyPort().get());
+                logger.warn("Configuration with \"proxy_port\" is deprecated. Use \"http_proxy.port\" instead.");
+                HttpProxy httpProxy = task.getHttpProxy().get();
+                if (!httpProxy.getPort().isPresent()) {
+                    httpProxy.setPort(task.getProxyPort());
+                    task.setHttpProxy(Optional.of(httpProxy));
+                }
             }
 
-            if (task.getAccessKeyId().isPresent()) {
-                BasicAWSCredentials basicAWSCredentials = new BasicAWSCredentials(
-                        task.getAccessKeyId().get(), task.getSecretAccessKey().get());
-
-                client = new AmazonS3Client(basicAWSCredentials, config);
+            if (task.getHttpProxy().isPresent()) {
+                setHttpProxyInAwsClient(clientConfig, task.getHttpProxy().get());
             }
-            else {
-                // Use default credential provider chain.
-                client = new AmazonS3Client(config);
+
+            return clientConfig;
+        }
+
+        private void setHttpProxyInAwsClient(ClientConfiguration clientConfig, HttpProxy httpProxy) {
+            // host
+            clientConfig.setProxyHost(httpProxy.getHost());
+
+            // port
+            if (httpProxy.getPort().isPresent()) {
+                clientConfig.setProxyPort(httpProxy.getPort().get());
             }
 
-            if (task.getEndpoint().isPresent()) {
-                client.setEndpoint(task.getEndpoint().get());
+            // https
+            clientConfig.setProtocol(httpProxy.getHttps() ? Protocol.HTTPS : Protocol.HTTP);
+
+            // user
+            if (httpProxy.getUser().isPresent()) {
+                clientConfig.setProxyUsername(httpProxy.getUser().get());
             }
 
-            return client;
+            // password
+            if (httpProxy.getPassword().isPresent()) {
+                clientConfig.setProxyPassword(httpProxy.getPassword().get());
+            }
         }
 
         public S3FileOutput(PluginTask task, int taskIndex)
