add an assume role provider (#150)

* add an assume role provider
* Add `assume-role` to copier options.  Use the role in the AssumeRoleCredentialProvider

Author: barnharts4

CHANGELOG.md

@@ -3,9 +3,7 @@
 ### Added
 * Table transformation to add custom properties to tables during a replication.
 * If a user doesn't specify `avro-serde-options`, Circus Train will still copy the external schema over to the target table. See [#131](https://github.com/HotelsDotCom/circus-train/issues/131).
-
-### Changed
-* Updated `jackson` version to 2.9.10 (was 2.9.9).
+* Added `copier-options.assume-role` to assume a role when using the S3MapReduceCp copier class. See [README.md](https://github.com/HotelsDotCom/circus-train) for details.
 
 ### Removed
 * Excluded `org.pentaho:pentaho-aggdesigner-algorithm` from build.
@@ -14,7 +12,7 @@
 * Bug in `AbstractAvroSerDeTransformation` where the config state wasn't refreshed on every replication.
 
 ### Changed
-* Updated `jackson` version to 2.9.9 (was 2.9.8).
+* Updated `jackson` version to 2.9.10 (was 2.9.8).
 * Updated `beeju` version to 2.0.0 (was 1.2.1).
 * Updated `circus-train-minimal.yml.template` to include the required `housekeeping` configuration for using the default schema with H2. 
 

README.md

@@ -376,6 +376,7 @@ If data is being replicated from HDFS to S3 then Circus Train will use a customi
 | `copier-options.upload-buffer-size`|No|Size of the buffer used to upload the stream of data. If the value is `0` the upload will use the value of the HDFS property `io.file.buffer.size` to configure the buffer. Defaults to `0`|
 | `copier-options.canned-acl`|No|AWS Canned ACL name. See [Access Control List (ACL) Overview](https://docs.aws.amazon.com/AmazonS3/latest/dev/acl-overview.html#canned-acl) for possible values. If not specified `S3MapReduceCp` will not specify any canned ACL.|
 | `copier-options.copier-factory-class`|No|Controls which copier is used for replication if provided.|
+| `copier-options.assume-role`|No|ARN of an IAM role to assume when writing S3 data to the target replica. Useful when the target is in a different AWS account than CircusTrain is running in. Note that if JCEKS is also configured, JCEKS credentials will be used instead of assuming a role. If `assume-role` is not specified, the copier will use instance credentials.|
 
 ##### S3 to S3 copier options
 If data is being replicated from S3 to S3 then Circus Train will use the AWS S3 API to copy data between S3 buckets. Using the AWS provided APIs no data needs to be downloaded or uploaded to the machine on which Circus Train is running but is copied by AWS internal infrastructure and stays in the AWS network boundaries. Assuming the correct bucket policies are in place cross region and cross account replication is supported. We are using the [TransferManager](https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/s3/transfer/TransferManager.html) to do the copying and we expose its options via copier-options see the table below. Given the source and target buckets Circus-Train will try to infer the region from them.
@@ -391,7 +392,14 @@ If data is being replicated from S3 to S3 then Circus Train will use the AWS S3
 |`copier-options.s3s3-retry-max-copy-attempts`|No|Controls the maximum number of attempts if AWS throws an error during copy. Default value is 3.|
 
 ### S3 Secret Configuration
-When configuring a job for replication to or from S3, the AWS access key and secret key with read/write access to the configured S3 buckets must be supplied. To protect these from being exposed in the job's Hadoop configuration, Circus Train expects them to be stored using the Hadoop Credential Provider and the JCEKS URL provided in the Circus Train configuration `security.credential-provider` property. This property is only required if a specific set of credentials is needed or if Circus Train runs on a non-AWS environment. If it is not set then the credentials of the instance where Circus Train runs will be used - note this scenario is only valid when Circus Train is executed on an AWS environment, i.e. EC2/EMR instance.
+When configuring a job for replication to or from S3, the AWS access key and secret key with read/write access to the configured S3 buckets must be supplied. Circus train has a couple of options depending on where you run Circus Train.
+* Running on EMR:
+We provide a `com.hotels.bdp.circustrain.aws.HadoopAWSCredentialProviderChain` that uses Jceks (explained below), STS assume role credentials (see `copier-options.assume-role`) and Instance Profile credentials. If the EMR cluster is running with a role that has read write access to the necessary buckets no configuration is necessary and instance profile credentials will be used.
+* Running on on-premises:
+It is possible to provide s3 secret configuration with jceks (explained below) or STS assume role credentials (see `copier-options.assume-role`).
+
+#### S3 Secret Configuration with Jceks
+The AWS access key and secret key can be protected from being exposed in the job's Hadoop configuration, Circus Train expects them to be stored using the Hadoop Credential Provider and the JCEKS URL provided in the Circus Train configuration `security.credential-provider` property. This property is only required if a specific set of credentials is needed or if Circus Train runs on a non-AWS environment. If it is not set then the credentials of the instance where Circus Train runs will be used - note this scenario is only valid when Circus Train is executed on an AWS environment, i.e. EC2/EMR instance.
 
 To add your existing AWS keys for a new replication job run the following commands as the user that will be executing Circus Train and pass in your keys when prompted:
 

circus-train-aws/pom.xml

@@ -29,6 +29,11 @@
       <groupId>org.springframework</groupId>
       <artifactId>spring-context</artifactId>
     </dependency>
+    <dependency>
+      <groupId>com.amazonaws</groupId>
+      <artifactId>aws-java-sdk-sts</artifactId>
+      <version>${aws-jdk.version}</version>
+    </dependency>
 
     <!-- Test -->
     <dependency>

circus-train-aws/src/main/java/com/hotels/bdp/circustrain/aws/AssumeRoleCredentialProvider.java

@@ -0,0 +1,60 @@
+/**
+ * Copyright (C) 2016-2019 Expedia Inc.
+ *
+ * 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 com.hotels.bdp.circustrain.aws;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.auth.STSAssumeRoleSessionCredentialsProvider;
+import com.amazonaws.auth.STSAssumeRoleSessionCredentialsProvider.Builder;
+
+public class AssumeRoleCredentialProvider implements AWSCredentialsProvider {
+
+  public static final String ASSUME_ROLE_PROPERTY_NAME = "com.hotels.bdp.circustrain.aws.AssumeRoleCredentialProvider.assumeRole";
+  private static final int CREDENTIALS_DURATION = 12 * 60 * 60; // max duration for assumed role credentials
+
+  private AWSCredentials credentials;
+  private final Configuration conf;
+
+  public AssumeRoleCredentialProvider(Configuration conf) {
+    this.conf = conf;
+  }
+
+  @Override
+  public AWSCredentials getCredentials() {
+    if (credentials == null) {
+      refresh();
+    }
+    return credentials;
+  }
+
+  @Override
+  public void refresh() {
+    checkNotNull(conf, "conf is required");
+    String roleArn = conf.get(ASSUME_ROLE_PROPERTY_NAME);
+    checkArgument(StringUtils.isNotEmpty(roleArn),
+        "Role ARN must not be empty, please set: " + ASSUME_ROLE_PROPERTY_NAME);
+
+    Builder builder = new STSAssumeRoleSessionCredentialsProvider.Builder(roleArn, "ct-assume-role-session");
+    credentials = builder.withRoleSessionDurationSeconds(CREDENTIALS_DURATION).build().getCredentials();
+  }
+
+}

circus-train-aws/src/main/java/com/hotels/bdp/circustrain/aws/HadoopAWSCredentialProviderChain.java

@@ -25,6 +25,7 @@
  * <ul>
  * <li>Credentials from Hadoop configuration set via JCE KS - if a JCE KS path or Hadoop {@code Configuration} is
  * provided</li>
+ * <li>{@link AssumeRoleCredentialProvider} that provides credentials by assuming a role</li>
  * <li>{@link EC2ContainerCredentialsProviderWrapper} that loads credentials from an Amazon Container (e.g. EC2)</li>
  * </ul>
  *
@@ -42,7 +43,8 @@ public HadoopAWSCredentialProviderChain(String credentialProviderPath) {
   }
 
   public HadoopAWSCredentialProviderChain(Configuration conf) {
-    super(new JceksAWSCredentialProvider(conf), new EC2ContainerCredentialsProviderWrapper());
+    // note that the order of these providers is significant as they will be tried in the order passed in
+    super(new JceksAWSCredentialProvider(conf), new AssumeRoleCredentialProvider(conf),
+        new EC2ContainerCredentialsProviderWrapper());
   }
-
 }

circus-train-aws/src/test/java/com/hotels/bdp/circustrain/aws/AssumeRoleCredentialProviderTest.java

@@ -0,0 +1,47 @@
+/**
+ * Copyright (C) 2016-2019 Expedia Inc.
+ *
+ * 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 com.hotels.bdp.circustrain.aws;
+
+import org.apache.hadoop.conf.Configuration;
+import org.junit.Test;
+
+public class AssumeRoleCredentialProviderTest {
+
+  @Test(expected = NullPointerException.class)
+  public void getCredentialsThrowsNullPointerException() {
+    AssumeRoleCredentialProvider provider = new AssumeRoleCredentialProvider(null);
+    provider.getCredentials();
+  }
+
+  @Test(expected = NullPointerException.class)
+  public void refreshThrowsNullPointerException() {
+    AssumeRoleCredentialProvider provider = new AssumeRoleCredentialProvider(null);
+    provider.refresh();
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void getCredentialsThrowsIllegalArgumentException() {
+    AssumeRoleCredentialProvider provider = new AssumeRoleCredentialProvider(new Configuration());
+    provider.getCredentials();
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void refreshThrowsIllegalArgumentException() {
+    AssumeRoleCredentialProvider provider = new AssumeRoleCredentialProvider(new Configuration());
+    provider.refresh();
+  }
+  
+}

circus-train-s3-mapreduce-cp-copier/src/main/java/com/hotels/bdp/circustrain/s3mapreducecpcopier/S3MapReduceCpOptionsParser.java

@@ -49,6 +49,7 @@ public class S3MapReduceCpOptionsParser {
   public static final String UPLOAD_RETRY_DELAY_MS = "upload-retry-delay-ms";
   public static final String UPLOAD_BUFFER_SIZE = "upload-buffer-size";
   public static final String CANNED_ACL = "canned-acl";
+  public static final String ASSUME_ROLE = "assume-role";
 
   private final S3MapReduceCpOptions.Builder optionsBuilder;
   private final URI defaultCredentialsProvider;
@@ -147,6 +148,8 @@ protected S3MapReduceCpOptions parse(Map<String, Object> copierOptions) {
 
     optionsBuilder.cannedAcl(MapUtils.getString(copierOptions, CANNED_ACL, ConfigurationVariable.CANNED_ACL.defaultValue()));
 
+    optionsBuilder.assumeRole(MapUtils.getString(copierOptions, ASSUME_ROLE, ConfigurationVariable.ASSUME_ROLE.defaultValue()));
+
     return optionsBuilder.build();
   }
 

circus-train-s3-mapreduce-cp/src/main/java/com/hotels/bdp/circustrain/s3mapreducecp/ConfigurationVariable.java

@@ -24,6 +24,8 @@
 import com.amazonaws.services.s3.model.StorageClass;
 import com.amazonaws.services.s3.transfer.TransferManagerConfiguration;
 
+import com.hotels.bdp.circustrain.aws.AssumeRoleCredentialProvider;
+
 final class Constants {
   static final TransferManagerConfiguration DEFAULT_TRANSFER_MANAGER_CONFIGURATION = new TransferManagerConfiguration();
 
@@ -32,6 +34,7 @@ private Constants() {}
 
 public enum ConfigurationVariable {
 
+  ASSUME_ROLE(AssumeRoleCredentialProvider.ASSUME_ROLE_PROPERTY_NAME, null),
   CANNED_ACL("com.hotels.bdp.circustrain.s3mapreducecp.cannedAcl", null),
   CREDENTIAL_PROVIDER("com.hotels.bdp.circustrain.s3mapreducecp.credentialsProvider", null),
   MINIMUM_UPLOAD_PART_SIZE("com.hotels.bdp.circustrain.s3mapreducecp.minimumUploadPartSize",

circus-train-s3-mapreduce-cp/src/main/java/com/hotels/bdp/circustrain/s3mapreducecp/S3MapReduceCpOptions.java

@@ -141,6 +141,11 @@ public Builder cannedAcl(String cannedAcl) {
       return this;
     }
 
+    public Builder assumeRole(String assumeRole) {
+      options.setAssumeRole(assumeRole);
+      return this;
+    }
+
     public S3MapReduceCpOptions build() {
       return options;
     }
@@ -213,6 +218,9 @@ public static Builder builder(List<Path> sources, URI target) {
   @Parameter(names = "--cannedAcl", description = "AWS Canned ACL")
   private String cannedAcl = ConfigurationVariable.CANNED_ACL.defaultValue();
 
+  @Parameter(names = "--assumeRole", description = "AWS IAM role to assume for writing to replica S3 bucket")
+  private String assumeRole = ConfigurationVariable.ASSUME_ROLE.defaultValue();
+
   public S3MapReduceCpOptions() {}
 
   public S3MapReduceCpOptions(S3MapReduceCpOptions options) {
@@ -237,6 +245,7 @@ public S3MapReduceCpOptions(S3MapReduceCpOptions options) {
     uploadRetryDelayMs = options.uploadRetryDelayMs;
     uploadBufferSize = options.uploadBufferSize;
     cannedAcl = options.cannedAcl;
+    assumeRole = options.assumeRole;
   }
 
   public boolean isHelp() {
@@ -411,6 +420,12 @@ public void setCannedAcl(String cannedAcl) {
     this.cannedAcl = cannedAcl;
   }
 
+  public String getAssumeRole() { return assumeRole; }
+
+  public void setAssumeRole(String assumeRole) {
+    this.assumeRole = assumeRole;
+  }
+
   public Map<String, String> toMap() {
     ImmutableMap.Builder<String, String> builder = ImmutableMap
         .<String, String>builder()
@@ -438,6 +453,9 @@ public Map<String, String> toMap() {
     if (cannedAcl != null) {
       builder.put(ConfigurationVariable.CANNED_ACL.getName(), cannedAcl);
     }
+    if (assumeRole != null) {
+      builder.put(ConfigurationVariable.ASSUME_ROLE.getName(), assumeRole);
+    }
     return builder.build();
   }
 
@@ -465,6 +483,7 @@ public String toString() {
         ", uploadRetryDelayMs=" + uploadRetryDelayMs +
         ", uploadBufferSize=" + uploadBufferSize +
         ", cannedAcl='" + cannedAcl + '\'' +
+        ", assumeRole='" + assumeRole + '\'' +
         '}';
   }
 }