github.com/aws/aws-msk-iam-auth @v2.3.7

Amazon MSK Library for AWS Identity and Access Management

License

This project is licensed under the Apache-2.0 License.

Introduction

The Amazon MSK Library for AWS Identity and Access Management enables developers to use AWS Identity and Access Management (IAM) to connect to their Amazon Managed Streaming for Apache Kafka (Amazon MSK) clusters. It allows JVM based Apache Kafka clients to use AWS IAM for authentication and authorization against Amazon MSK clusters that have AWS IAM enabled as an authentication mechanism.

This library provides a new Simple Authentication and Security Layer (SASL) mechanism called AWS_MSK_IAM. This new SASL mechanism can be used by Kafka clients to authenticate against Amazon MSK clusters using AWS IAM.

• [Amazon Managed Streaming for Apache Kafka][MSK]
• [AWS Identity and Access Management][IAM]
• [AWS IAM authentication and authorization for MSK ][MSK_IAM]

Building from source

After you've downloaded the code from GitHub, you can build it using Gradle. Use this command:

gradle clean build

The generated jar files can be found at: build/libs/.

An uber jar containing the library and all its relocated dependencies except the kafka client and slf4j-api can also be built. Use this command:

gradle clean shadowJar

The generated uber jar file can also be found at: build/libs/. At runtime, the uber jar expects to find the kafka client library and the sl4j-api library on the classpath.

Validating secure dependencies

To ensure no security vulnerabilities in the dependency libraries, run the following.

gradle dependencyCheckAnalyze

If the above reports any vulnerabilities, upgrade dependencies to use the respective latest versions.

Using the Amazon MSK Library for IAM Authentication

The recommended way to use this library is to consume it from maven central while building a Kafka client application.

xml <dependency> <groupId>software.amazon.msk</groupId> <artifactId>aws-msk-iam-auth</artifactId> <version>2.3.6</version> </dependency> If you want to use it with a pre-existing Kafka client, you could build the uber jar and place it in the Kafka client's classpath.

Configuring a Kafka client to use AWS IAM with AWS_MSK_IAM mechanism

You can configure a Kafka client to use AWS IAM for authentication by adding the following properties to the client's configuration.

# Sets up TLS for encryption and SASL for authN.
security.protocol = SASL_SSL

# Identifies the SASL mechanism to use.
sasl.mechanism = AWS_MSK_IAM

# Binds SASL client implementation.
sasl.jaas.config = software.amazon.msk.auth.iam.IAMLoginModule required;

# Encapsulates constructing a SigV4 signature based on extracted credentials.
# The SASL client bound by "sasl.jaas.config" invokes this class.
sasl.client.callback.handler.class = software.amazon.msk.auth.iam.IAMClientCallbackHandler

Configuring a Kafka client to use AWS IAM with SASL OAUTHBEARER mechanism

You can alternatively use SASL/OAUTHBEARER mechanism using IAM authentication by adding following configuration. For more details on SASL/OAUTHBEARER mechanism, please read - KIP-255

# Sets up TLS for encryption and SASL for authN.
security.protocol=SASL_SSL
# Identifies the SASL mechanism to use.
sasl.mechanism=OAUTHBEARER
# Binds SASL client implementation. You can add client credential configurations here.
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;
# Encapsulates constructing a SigV4 signature based on extracted credentials.
# The SASL client bound by "sasl.jaas.config" invokes this class.
sasl.login.callback.handler.class=software.amazon.msk.auth.iam.IAMOAuthBearerLoginCallbackHandler
# This is used during client authentication and reauthentication
sasl.client.callback.handler.class=software.amazon.msk.auth.iam.IAMOAuthBearerLoginCallbackHandler

This configuration finds IAM credentials using the [AWS Default Credentials Provider Chain][DefaultCreds]. To summarize, the Default Credential Provider Chain looks for credentials in this order:

1. Environment variables: AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY.
2. Java system properties: aws.accessKeyId and aws.secretKey.
3. Web Identity Token credentials from the environment or container.
4. The default credential profiles file– typically located at ~/.aws/credentials (location can vary per platform), and shared by many of the AWS SDKs and by the AWS CLI.
   You can create a credentials file by using the aws configure command provided by the AWS CLI, or you can create it by editing the file with a text editor. For information about the credentials file format, see [AWS Credentials File Format][CredsFile].
5. It can be used to load credentials from credential profiles other than the default one by setting the environment variable
   AWS_PROFILE to the name of the alternate credential profile. Profiles can be used to load credentials from other sources such as AWS IAM Roles. See [AWS Credentials File Format][CredsFile] for more details.
6. Amazon ECS container credentials– loaded from the Amazon ECS if the environment variable AWS_CONTAINER_CREDENTIALS_RELATIVE_URI is set.
7. Instance profile credentials: used on EC2 instances, and delivered through the Amazon EC2 metadata service.

Specifying an alternate credential profile for a client

If the client wants to specify a particular credential profile as part of the client configuration rather than through the environment variable AWS_PROFILE, they can pass in the name of the profile as a client configuration property:

# Binds SASL client implementation. Uses the specified profile name to look for credentials.
sasl.jaas.config = software.amazon.msk.auth.iam.IAMLoginModule required awsProfileName="<Credential Profile Name>";

Specifying a role based credential profile for a client

Some clients may want to assume a role and use the role's temporary credentials to communicate with a MSK cluster. One way to do that is to create a credential profile for that role following the rules for [Using an IAM role in the CLI][RoleProfileCLI]. They can then pass in the name of the credential profile as described above.

As an example, let's say a Kafka client is running on an Ec2 instance and the Kafka client wants to use an IAM role called msk_client_role. The Ec2 instance profile has permissions to assume the msk_client_role IAM role although msk_client_role is not attached to the instance profile.

In such a case, we create a credential profile called msk_client that assumes the role msk_client_role. The credential profile looks like:

[msk_client]
role_arn = arn:aws:iam::123456789012:role/msk_client_role
credential_source = Ec2InstanceMetadata

The credential profile name msk_client is passed in as a client configuration property:

sasl.jaas.config = software.amazon.msk.auth.iam.IAMLoginModule required awsProfileName="msk_client";

Many more examples of configuring credential profiles with IAM roles can be found in [Using an IAM role in the CLI][RoleProfileCLI].

Specifying an AWS IAM Role for a client

The library supports another way to configure a client to assume an IAM role and use the role's temporary credentials. The IAM role's ARN and optionally the session name for the client can be passed in as client configuration property:

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";

In this case, the awsRoleArn specifies the ARN for the IAM role the client should use and awsRoleSessionName specifies the session name that this particular client should use while assuming the IAM role. If the same IAM Role is used in multiple contexts, the session names can be used to differentiate between the different contexts. The awsRoleSessionName is optional.

awsStsRegion optionally specifies the regional endpoint of AWS STS to use while assuming the IAM role. If awsStsRegion is omitted the global endpoint for AWS STS is used by default. When the Kafka client is running in a VPC with an [STS interface VPC Endpoint][StsVpcE] [(AWS PrivateLink)][PrivateLink] to a regional endpoint of AWS STS and we want all STS traffic to go over that endpoint, we should set awsStsRegion to the region corresponding to the interface VPC Endpoint. It may also be necessary to configure the sts_regional_endpoints shared AWS config file setting, or the AWS_STS_REGIONAL_ENDPOINTS environment variable as per the [AWS STS Regionalized endpoints documentation.][StsRegionalEndpointsDoc]

The Default Credential Provider Chain must contain the permissions necessary to assume the client role. For example, if the client is an EC2 instance, its instance profile should have permission to assume the msk_client_role.

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 provider fails to obtain credentials, the fallback mechanism will use the default credential chain. To avoid this, set the awsAddDefaultProviders parameter to false (if not set, the default value is true):

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. So, in version 1.1.3 the library retries loading the credentials when it gets an SdkClientException (which wraps most AWS SDK client side exceptions). Since the retries do not impact the fault-free path and we had heard of user issues around random failures loading credentials (e.g.: #59, maybe #51 ), we decided to change the default behavior to retry a maximum of 3 times. It exponentially backs off with full jitter upto a max-delay of 2000 ms.

The maximum number of retries and the maximum back off period can be set:

sasl.jaas.config=software.amazon.msk.auth.iam.IAMLoginModule required awsMaxRetries="7" awsMaxBackOffTimeMs="500";

This sets the maximum number of retries to 7 and the maximum back off time to 500 ms.

The retries can be turned off completely by setting awsMaxRetries to "0".

Custom Region Provider

By default, the library extracts the AWS region from the MSK broker hostname. If the region cannot be determined from the hostname, it falls back to the DefaultAwsRegionProviderChain.

You can override this behavior by specifying a custom region provider via the awsMskRegionProvider JAAS config option. The value is a descriptor string in the format className?param1=value1;param2=value2. The referenced class must implement the ConfigurableRegionProvider interface and provide a constructor that accepts a Map<String, String>.

sasl.jaas.config=software.amazon.msk.auth.iam.IAMLoginModule required \
  awsMskRegionProvider="software.amazon.msk.auth.iam.internals.region.Route53RegionProvider?host=region.my-cluster.example.com;refresh.seconds=60";

The library ships with Route53RegionProvider, which resolves the region by performing a DNS TXT record lookup. It accepts the following optional parameters: - host — when provided, it is used directly as the DNS lookup name. When omitted, the broker hostname is prefixed with region. to form the lookup name (e.g. region.broker.example.com). - refresh.seconds — cache TTL in seconds for the resolved region. Defaults to 60 (1 minute). Set to 0 to disable caching and resolve DNS on every call.

The TXT record value is expected to contain a valid AWS region id (e.g. us-east-1).

If the custom region provider returns null or throws an exception, a warning is logged and the library falls back to the DefaultAwsRegionProviderChain.

You can also implement your own provider by implementing the ConfigurableRegionProvider interface:

package com.example;

import java.util.Map;
import software.amazon.awssdk.regions.Region;
import software.amazon.msk.auth.iam.internals.region.ConfigurableRegionProvider;

public class MyRegionProvider implements ConfigurableRegionProvider {
    public MyRegionProvider(Map<String, String> config) {
        // initialize from config params
    }

    @Override
    public Region getRegion() {
        return getRegion(null);
    }

    @Override
    public Region getRegion(String host) {
        return Region.of("eu-west-1");
    }
}

Then reference it in your JAAS config:

sasl.jaas.config=software.amazon.msk.auth.iam.IAMLoginModule required \
  awsMskRegionProvider="com.example.MyRegionProvider?myParam=myValue";

Setting EKS Service Account

If your Kafka Client, Producer or Consumer, is ru