add awsAddDefaultProviders flag

Author: Ed Berezitsky

README.md

@@ -159,13 +159,21 @@ The Default Credential Provider Chain must contain the permissions necessary to
 For example, if the client is an EC2 instance, its instance profile should have permission to assume the
  `msk_client_role`.
 
-When assume role method fails, the library will use fallback strategy to try other providers from the default credential providers chain.
-To avoid this, use `skipCredChain="true"`. This will enable retry mechanism only for `STSAssumeRoleCredentialProvider`.
-
 ### Figuring out whether or not to use default credentials
 
 When you want the MSK client to connect to MSK using credentials not found in the [AWS Default Credentials Provider Chain][DefaultCreds], you can specify an `awsProfileName` containing the credential profile to use, or an `awsRoleArn` to indicate an IAM Role’s ARN to assume using credentials in the Default Credential Provider Chain. These parameters are optional, and if they are not set the MSK client will use credentials from the Default Credential Provider Chain. There is no need to specify them if you intend to use an IAM role associated with an AWS compute service, such as EC2 or ECS to authenticate to MSK.
 
+If Assume Role or Profile Name params are set, but a providers fails to obtain credentials, the fallback mechanism will use default credential chain.
+To avoid this, set `awsAddDefaultProviders` parameter to `false` (if not set, the default value is `true`):
+
+```properties
+sasl.jaas.config=software.amazon.msk.auth.iam.IAMLoginModule required \
+ awsRoleArn="arn:aws:iam::123456789012:role/msk_client_role" \
+ awsRoleSessionName="producer" \
+ awsStsRegion="us-west-2" \
+ awsAddDefaultProviders="false";
+```
+
 ### Retries while getting credentials
 In some scenarios the IAM credentials might be transiently unavailable. This will cause the connection to fail, which
 might in some cases cause the client application to stop. 

src/main/java/software/amazon/msk/auth/iam/internals/MSKCredentialProvider.java

@@ -89,7 +89,7 @@ public class MSKCredentialProvider implements AwsCredentialsProvider, AutoClosea
  private static final String AWS_ROLE_SESSION_KEY = "awsRoleSessionName";
  private static final String AWS_ROLE_SESSION_TOKEN = "awsRoleSessionToken";
  private static final String AWS_STS_REGION = "awsStsRegion";
- private static final String AWS_SKIP_CRED_CHAIN = "skipCredChain";
+ private static final String AWS_ADD_DEFAULT_PROVIDERS = "awsAddDefaultProviders";
  private static final String AWS_DEBUG_CREDS_KEY = "awsDebugCreds";
  private static final String AWS_SHOULD_USE_FIPS = "awsShouldUseFips";
  private static final String AWS_MAX_RETRIES = "awsMaxRetries";
@@ -112,26 +112,26 @@ public MSKCredentialProvider(Map<String, ?> options) {
 
  MSKCredentialProvider(ProviderBuilder builder) {
  this(builder.getProviders(), builder.shouldDebugCreds(), builder.getStsRegion(), builder.getMaxRetries(),
- builder.getMaxBackOffTimeMs(), builder.skipCredChain());
+ builder.getMaxBackOffTimeMs(), builder.addDefaultProviders());
  }
 
  MSKCredentialProvider(List<AwsCredentialsProvider> providers,
  Boolean shouldDebugCreds,
  String stsRegion,
  int maxRetries,
  int maxBackOffTimeMs) {
- this(providers, shouldDebugCreds, stsRegion, maxRetries, maxBackOffTimeMs, false);
+ this(providers, shouldDebugCreds, stsRegion, maxRetries, maxBackOffTimeMs, true);
  }
 
  MSKCredentialProvider(List<AwsCredentialsProvider> providers,
  Boolean shouldDebugCreds,
  String stsRegion,
  int maxRetries,
  int maxBackOffTimeMs,
- boolean skipCredChain) {
+ boolean addDefaultProviders) {
  AwsCredentialsProviderChain.Builder chain = AwsCredentialsProviderChain.builder();
  chain.credentialsProviders(providers);
- if (!skipCredChain) {
+ if (addDefaultProviders) {
  chain.addCredentialsProvider(getDefaultProvider());
  }
  compositeDelegate = chain.build();
@@ -273,8 +273,8 @@ public List<AwsCredentialsProvider> getProviders() {
  return providers;
  }
 
- public Boolean skipCredChain() {
- return Optional.ofNullable(optionsMap.get(AWS_SKIP_CRED_CHAIN)).map(d -> d.equals("true")).orElse(false);
+ public Boolean addDefaultProviders() {
+ return Optional.ofNullable(optionsMap.get(AWS_ADD_DEFAULT_PROVIDERS)).map(d -> d.equals("true")).orElse(true);
  }
 
  public Boolean shouldDebugCreds() {

src/test/java/software/amazon/msk/auth/iam/internals/MSKCredentialProviderTest.java

@@ -74,7 +74,7 @@ public class MSKCredentialProviderTest {
  private static final String AWS_ROLE_SECRET_ACCESS_KEY = "awsRoleSecretAccessKey";
  private static final String AWS_PROFILE_NAME = "awsProfileName";
  private static final String AWS_DEBUG_CREDS_NAME = "awsDebugCreds";
- private static final String AWS_SKIP_CRED_CHAIN = "skipCredChain";
+ private static final String AWS_ADD_DEFAULT_PROVIDERS = "awsAddDefaultProviders";
 
  /**
  * If no options are passed in it should use the default credentials provider
@@ -784,46 +784,40 @@ private URL getProfileResourceURL() {
  }
 
  @Test
- public void testSkipCredChainTrue() {
- Map<String, String> optionsMap = new HashMap<>();
- optionsMap.put(AWS_SKIP_CRED_CHAIN, "true");
- optionsMap.put(AWS_PROFILE_NAME, "profile-1");
-
- 
- MSKCredentialProvider provider = new MSKCredentialProvider(optionsMap) {
- protected AwsCredentialsProvider getDefaultProvider() {
- throw new RuntimeException("Default provider should not be called when skipCredChain is true");
- }
- };
- 
- assertThrows(SdkClientException.class, () -> provider.resolveCredentials());
- }
-
- @Test
- public void testSkipCredChainFalse() {
+ public void testAddDefaultProvidersTrue() {
  runTestWithSystemPropertyCredentials(() -> {
  Map<String, String> optionsMap = new HashMap<>();
- optionsMap.put(AWS_SKIP_CRED_CHAIN, "false");
+ optionsMap.put(AWS_PROFILE_NAME, "profile-1");
+ optionsMap.put(AWS_ADD_DEFAULT_PROVIDERS, "true");
+
  MSKCredentialProvider provider = new MSKCredentialProvider(optionsMap);
- 
  AwsCredentials credentials = provider.resolveCredentials();
- 
+
  assertEquals(ACCESS_KEY_VALUE, credentials.accessKeyId());
  assertEquals(SECRET_KEY_VALUE, credentials.secretAccessKey());
+
  }, ACCESS_KEY_VALUE, SECRET_KEY_VALUE);
  }
 
  @Test
- public void testSkipCredChainNotSet() {
+ public void testAddDefaultProvidersFalse() {
+ Map<String, String> optionsMap = new HashMap<>();
+ optionsMap.put(AWS_ADD_DEFAULT_PROVIDERS, "false");
+ // if MSKCredentialProviders is configured with an empty chain of providers, it should throw IllegalArgumentException
+ assertThrows(IllegalArgumentException.class, () -> new MSKCredentialProvider(optionsMap));
+ }
+
+ @Test
+ public void testAddDefaultProvidersNotSet() {
  runTestWithSystemPropertyCredentials(() -> {
  Map<String, String> optionsMap = new HashMap<>();
+
  MSKCredentialProvider provider = new MSKCredentialProvider(optionsMap);
- 
  AwsCredentials credentials = provider.resolveCredentials();
- 
+
  assertEquals(ACCESS_KEY_VALUE, credentials.accessKeyId());
  assertEquals(SECRET_KEY_VALUE, credentials.secretAccessKey());
- }, ACCESS_KEY_VALUE, SECRET_KEY_VALUE);
- }
+
+  }, ACCESS_KEY_VALUE, SECRET_KEY_VALUE); }
 
 }