feat: SASL client (for JAAS-compatible apps)

Add a basic SASL client using the AWS SDK which will leverage the
DefaultAWSCredentialsProviderChain to automatically discover and
use the appropriate AWS credentials based on your environment
variables, AWS credentials files or available metadata sources
(EC2, ECS, etc.)

README.md

@@ -3,7 +3,7 @@
 This is a custom LoginModule meant to be used with [Kafka](https://kafka.apache.org) and configured
 via [JAAS](https://docs.oracle.com/javase/8/docs/technotes/guides/security/jaas/JAASRefGuide.html).
 
-## Usage
+## Server Setup
 
 Copy the `kafka-auth-aws-iam-{VERSION}.jar` file into your Kafka's `libs` directory.
 Next, create a JAAS configuration file that looks something like this:
@@ -38,3 +38,37 @@ Without this entry, the `SaslServerCallbackhandler` will be used, which will fai
 # So, for SASL_SSL, it should be:
 listener.name.sasl_ssl.aws.sasl.server.callback.handler.class=com.stack.security.auth.aws.internal.AwsIamCallbackHandler
 ```
+
+## Client Setup
+
+You can use the AwsIamLoginModule for authentication between Kafka Brokers. To do this,
+add an entry to your JAAS configuration for `KafkaClient`:
+
+```
+KafkaClient {
+    com.stack.security.auth.aws.AwsIamLoginModule required;
+};
+```
+
+The `AwsIamSaslClient` leverages the [DefaultAWSCredentialsProviderChain](https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/auth/DefaultAWSCredentialsProviderChain.html) to automatically find credentials available for use in
+the environment of your Kafka broker.
+
+Unlike the `AwsIamSaslServer`, the `AwsIamSaslClient` does not use a custom callback so no
+additional wireup is needed in server.properties.
+
+This client can also be used with JAAS-compatible Consumers and Producers, though this hasn't been
+tested.
+
+
+## Build, Test, etc...
+This project uses [Maven](https://maven.apache.org/).
+```
+# Install dependencies
+mvn install
+# Build
+mvn compile
+# Test
+mvn test
+# Package the JAR for use
+mvn package
+```

pom.xml

@@ -5,7 +5,7 @@
 
   <groupId>com.stack.security.auth.aws</groupId>
   <artifactId>kafka-auth-aws-iam</artifactId>
-  <version>0.2.2</version>
+  <version>0.3.0</version>
   <packaging>jar</packaging>
   <properties>
     <maven.compiler.source>8</maven.compiler.source>
@@ -85,6 +85,7 @@
           <descriptorRefs>
             <descriptorRef>jar-with-dependencies</descriptorRef>
           </descriptorRefs>
+          <appendAssemblyId>false</appendAssemblyId>
         </configuration>
         <executions>
           <execution>

src/main/java/com/stack/security/auth/aws/AwsIamAuthenticateCallback.java

@@ -24,33 +24,46 @@ public class AwsIamAuthenticateCallback implements Callback {
    *                        SASL/AWS authentication
    * @return
    */
-  public AwsIamAuthenticateCallback(char[] accessKeyId, char[] secretAccessKey, char[] sessionToken) {
-    this.accessKeyId = accessKeyId;
-    this.secretAccessKey = secretAccessKey;
-    this.sessionToken = sessionToken;
+  public AwsIamAuthenticateCallback(String accessKeyId, String secretAccessKey, String sessionToken) {
+    setAccessKeyId(accessKeyId);
+    setSecretAccessKey(secretAccessKey);
+    this.secretAccessKey = secretAccessKey.toCharArray();
+    this.sessionToken = sessionToken.toCharArray();
   }
 
   /**
    * Returns the AWS Access Key ID provided by the client during SASL/AWS
    */
-  public char[] accessKeyId() {
+  public char[] getAccessKeyId() {
     return accessKeyId;
   }
 
   /**
    * Returns the AWS Secret Access Key provided by the client during SASL/AWS
    */
-  public char[] secretAccessKey() {
+  public char[] getSecretAccessKey() {
     return secretAccessKey;
   }
 
   /**
    * Returns the AWS Session Token provided by the client during SASL/AWS
    */
-  public char[] sessionToken() {
+  public char[] getSessionToken() {
     return sessionToken;
   }
 
+  public void setAccessKeyId(String accessKeyId) {
+    this.accessKeyId = accessKeyId.toCharArray();
+  }
+
+  public void setSecretAccessKey(String secretAccessKey) {
+    this.secretAccessKey = secretAccessKey.toCharArray();
+  }
+
+  public void setSessionToken(String sessionToken) {
+    this.sessionToken = sessionToken.toCharArray();
+  }
+
   /**
    * Returns true if client password matches expected password, false otherwise.
    * This state is set by the server-side callback handler.

src/main/java/com/stack/security/auth/aws/AwsIamLoginModule.java

@@ -7,6 +7,7 @@
 import javax.security.auth.callback.CallbackHandler;
 import javax.security.auth.spi.LoginModule;
 
+import com.stack.security.auth.aws.internal.AwsIamSaslClientProvider;
 import com.stack.security.auth.aws.internal.AwsIamSaslServerProvider;
 
 public class AwsIamLoginModule implements LoginModule {
@@ -18,6 +19,7 @@ public class AwsIamLoginModule implements LoginModule {
 
   static {
     AwsIamSaslServerProvider.initialize();
+    AwsIamSaslClientProvider.initialize();
   }
 
   @Override

src/main/java/com/stack/security/auth/aws/internal/AwsIamClientCallbackHandler.java

@@ -0,0 +1,26 @@
+package com.stack.security.auth.aws.internal;
+
+import java.util.List;
+import java.util.Map;
+
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.UnsupportedCallbackException;
+import javax.security.auth.login.AppConfigurationEntry;
+
+import org.apache.kafka.common.security.auth.AuthenticateCallbackHandler;
+
+public class AwsIamClientCallbackHandler implements AuthenticateCallbackHandler {
+
+  @Override
+  public void configure(Map<String, ?> configs, String saslMechanism, List<AppConfigurationEntry> jaasConfigEntries) {
+  }
+
+  @Override
+  public void handle(Callback[] callbacks) throws UnsupportedCallbackException {
+
+  }
+
+  @Override
+  public void close() {
+  }
+}