Enable support for aws eks rbac and aws iam msk

pom.xml

@@ -18,6 +18,8 @@
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <protobuf.version>3.11.1</protobuf.version>
         <testcontainers.version>1.14.3</testcontainers.version>
+        <msk.auth.version>1.0.0</msk.auth.version>
+        <sts.sdk.version>1.11.704</sts.sdk.version>
     </properties>
 
     <scm>
@@ -86,6 +88,11 @@
             <artifactId>lombok</artifactId>
             <version>1.18.8</version>
         </dependency>
+        <dependency>
+            <groupId>software.amazon.msk</groupId>
+            <artifactId>aws-msk-iam-auth</artifactId>
+            <version>${msk.auth.version}</version>
+        </dependency>
         <dependency>
             <groupId>io.confluent</groupId>
             <artifactId>kafka-avro-serializer</artifactId>
@@ -154,6 +161,12 @@
             <artifactId>spring-kafka</artifactId>
             <version>2.6.7</version>
         </dependency>
+
+        <dependency>
+            <groupId>com.amazonaws</groupId>
+            <artifactId>aws-java-sdk-sts</artifactId>
+            <version>${sts.sdk.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-undertow</artifactId>

src/main/java/kafdrop/config/KafkaConfiguration.java

@@ -23,11 +23,17 @@ public final class KafkaConfiguration {
   private String truststoreFile;
   private String propertiesFile;
   private String keystoreFile;
+  private String jaasConfig;
+  private String clientCallback;
+  private String iamEnabled;
 
   public void applyCommon(Properties properties) {
     properties.setProperty(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokerConnect);
+
     if (isSecured) {
       LOG.warn("The 'isSecured' property is deprecated; consult README.md on the preferred way to configure security");
+      LOG.info("Setting security protocol to {}", securityProtocol);
+      LOG.info("Setting sasl mechanism to {}", saslMechanism);
       properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, securityProtocol);
       properties.put(SaslConfigs.SASL_MECHANISM, saslMechanism);
     }
@@ -37,6 +43,12 @@ public void applyCommon(Properties properties) {
       LOG.info("Assigning truststore location to {}", truststoreFile);
       properties.put("ssl.truststore.location", truststoreFile);
     }
+    LOG.info("Is iam enabled : {}", iamEnabled);
+    if (Boolean.parseBoolean(iamEnabled)) {
+      LOG.info("Setting sasl.jaas.config {} and sasl and callback callback properties {}", jaasConfig, clientCallback);
+      properties.put(SaslConfigs.SASL_CLIENT_CALLBACK_HANDLER_CLASS, clientCallback);
+      properties.put(SaslConfigs.SASL_JAAS_CONFIG, jaasConfig);
+    }
 
     LOG.info("Checking keystore file {}", keystoreFile);
     if (new File(keystoreFile).isFile()) {

src/main/resources/application.yml

@@ -28,9 +28,12 @@ kafdrop.monitor:
 
 kafka:
   brokerConnect: localhost:9092
-  isSecured: false
-  saslMechanism: "PLAIN"
-  securityProtocol: "SASL_PLAINTEXT"
+  isSecured: "${KAFKA_IS_SECURED:false}"
+  saslMechanism: "${KAFKA_SASL_MECHANISM:PLAIN}"
+  securityProtocol: "${KAFKA_SECURITY_PROTOCOL:SASL_PLAINTEXT}"
   truststoreFile: "${KAFKA_TRUSTSTORE_FILE:kafka.truststore.jks}"
   propertiesFile : "${KAFKA_PROPERTIES_FILE:kafka.properties}"
-  keystoreFile: "${KAFKA_KEYSTORE_FILE:kafka.keystore.jks}"
+  keystoreFile: "${KAFKA_KEYSTORE_FILE:kafka.keystore.jks}"
+  iamEnabled: "${KAFKA_IAM_ENABLED:false}"
+  jaasConfig: "${KAFKA_JAAS_CONFIG}"
+  clientCallback: "software.amazon.msk.auth.iam.IAMClientCallbackHandler"