Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Pass region into credentials provider #695

Closed
wants to merge 1 commit into from
+21 −19
Closed

Pass region into credentials provider #695

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions kafka-connect-s3/src/main/java/io/confluent/connect/s3/S3SinkConnectorConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -896,6 +896,7 @@ public AWSCredentialsProvider getCredentialsProvider() {

configs.put(AWS_ACCESS_KEY_ID_CONFIG, awsAccessKeyId());
configs.put(AWS_SECRET_ACCESS_KEY_CONFIG, awsSecretKeyId().value());
configs.put(REGION_CONFIG, getString(REGION_CONFIG));

((Configurable) provider).configure(configs);
} else {
Expand Down
34 changes: 17 additions & 17 deletions ...nnect-s3/src/main/java/io/confluent/connect/s3/auth/AwsAssumeRoleCredentialsProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@

import static io.confluent.connect.s3.S3SinkConnectorConfig.AWS_ACCESS_KEY_ID_CONFIG;
import static io.confluent.connect.s3.S3SinkConnectorConfig.AWS_SECRET_ACCESS_KEY_CONFIG;
import static io.confluent.connect.s3.S3SinkConnectorConfig.REGION_CONFIG;

/**
* AWS credentials provider that uses the AWS Security Token Service to assume a Role and create a
Expand Down Expand Up @@ -79,26 +80,25 @@ public void configure(Map<String, ?> configs) {
roleSessionName = config.getString(ROLE_SESSION_NAME_CONFIG);
final String accessKeyId = (String) configs.get(AWS_ACCESS_KEY_ID_CONFIG);
final String secretKey = (String) configs.get(AWS_SECRET_ACCESS_KEY_CONFIG);
final String region = (String) configs.get(REGION_CONFIG);

// default sts client will internally use default credentials chain provider
AWSSecurityTokenServiceClientBuilder stsClientBuilder = AWSSecurityTokenServiceClientBuilder
.standard()
.withRegion(region);

// Use explicit access key and secret if set
if (StringUtils.isNotBlank(accessKeyId) && StringUtils.isNotBlank(secretKey)) {
basicCredentials = new BasicAWSCredentials(accessKeyId, secretKey);
stsCredentialProvider = new STSAssumeRoleSessionCredentialsProvider
.Builder(roleArn, roleSessionName)
.withStsClient(AWSSecurityTokenServiceClientBuilder
.standard()
.withCredentials(new AWSStaticCredentialsProvider(basicCredentials)).build()
)
.withExternalId(roleExternalId)
.build();
} else {
basicCredentials = null;
stsCredentialProvider = new STSAssumeRoleSessionCredentialsProvider
.Builder(roleArn, roleSessionName)
// default sts client will internally use default credentials chain provider
// https://docs.aws.amazon.com/sdk-for-java/v1/developer-guide/credentials.html#credentials-default
.withStsClient(AWSSecurityTokenServiceClientBuilder.defaultClient())
.withExternalId(roleExternalId)
.build();
stsClientBuilder = stsClientBuilder
.withCredentials(new AWSStaticCredentialsProvider(basicCredentials));
}

stsCredentialProvider = new STSAssumeRoleSessionCredentialsProvider
.Builder(roleArn, roleSessionName)
.withStsClient(stsClientBuilder.build())
.withExternalId(roleExternalId)
.build();
}

@Override
Expand Down
3 changes: 2 additions & 1 deletion kafka-connect-s3/src/test/java/io/confluent/connect/s3/S3SinkConnectorConfigTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -226,7 +226,7 @@ public void testConfigurableCredentialProvider() {
);
properties.put(
configPrefix.concat(DummyAssertiveCredentialsProvider.CONFIGS_NUM_KEY_NAME),
"5"
"6"
);
connectorConfig = new S3SinkConnectorConfig(properties);

Expand Down Expand Up @@ -255,6 +255,7 @@ public void testConfigurableAwsAssumeRoleCredentialsProvider() {
configPrefix.concat(AwsAssumeRoleCredentialsProvider.ROLE_EXTERNAL_ID_CONFIG),
"my-external-id"
);
properties.put(S3SinkConnectorConfig.REGION_CONFIG, "us-west-2");
connectorConfig = new S3SinkConnectorConfig(properties);

AwsAssumeRoleCredentialsProvider credentialsProvider =
Expand Down
2 changes: 1 addition & 1 deletion kafka-connect-s3/src/test/java/io/confluent/connect/s3/storage/S3StorageTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -142,7 +142,7 @@ public void testUserDefinedCredentialsProvider() throws Exception {
String configPrefix = S3SinkConnectorConfig.CREDENTIALS_PROVIDER_CONFIG_PREFIX;
localProps.put(configPrefix.concat(DummyAssertiveCredentialsProvider.ACCESS_KEY_NAME), "foo_key");
localProps.put(configPrefix.concat(DummyAssertiveCredentialsProvider.SECRET_KEY_NAME), "bar_secret");
localProps.put(configPrefix.concat(DummyAssertiveCredentialsProvider.CONFIGS_NUM_KEY_NAME), "5");
localProps.put(configPrefix.concat(DummyAssertiveCredentialsProvider.CONFIGS_NUM_KEY_NAME), "6");
localProps.put(
S3SinkConnectorConfig.CREDENTIALS_PROVIDER_CLASS_CONFIG,
DummyAssertiveCredentialsProvider.class.getName()
Expand Down