JMX SSL and auth (#818)

* Secured JMX WIP

* Make credentials nullability more obvious

Add example keys/certs

* Add required jmxSsl option

* Update README.md

README.md

@@ -179,3 +179,6 @@ For example, if you want to use an environment variable to set the `name` parame
 |`LOGGING_LEVEL_ROOT`        	| Setting log level (all, debug, info, warn, error, fatal, off). Default: debug
 |`LOGGING_LEVEL_COM_PROVECTUS`        	|Setting log level (all, debug, info, warn, error, fatal, off). Default: debug
 |`SERVER_PORT` |Port for the embedded server. Default `8080`
+|`KAFKA_CLUSTERS_0_JMXSSL` |Enable SSL for JMX? `true` or `false`. For advanced setup see `kafka-ui-jmx-secured.yml`
+|`KAFKA_CLUSTERS_0_JMXUSERNAME` |Username for JMX authentication
+|`KAFKA_CLUSTERS_0_JMXPASSWORD` |Password for JMX authentication

docker/jmx/jmxremote.access

@@ -0,0 +1 @@
+root readwrite

docker/jmx/jmxremote.password

@@ -0,0 +1 @@
+root password

docker/kafka-ui-jmx-secured.yml

@@ -0,0 +1,136 @@
+---
+version: '2'
+services:
+
+  kafka-ui:
+    container_name: kafka-ui
+    image: provectuslabs/kafka-ui:latest
+    ports:
+      - 8080:8080
+      - 5005:5005
+    depends_on:
+      - zookeeper0
+      - kafka0
+      - schemaregistry0
+      - kafka-connect0
+    environment:
+      KAFKA_CLUSTERS_0_NAME: local
+      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka0:29092
+      KAFKA_CLUSTERS_0_ZOOKEEPER: zookeeper0:2181
+      KAFKA_CLUSTERS_0_SCHEMAREGISTRY: http://schemaregistry0:8085
+      KAFKA_CLUSTERS_0_KAFKACONNECT_0_NAME: first
+      KAFKA_CLUSTERS_0_KAFKACONNECT_0_ADDRESS: http://kafka-connect0:8083
+      KAFKA_CLUSTERS_0_JMXPORT: 9997
+      KAFKA_CLUSTERS_0_JMXSSL: 'true'
+      KAFKA_CLUSTERS_0_JMXUSERNAME: root
+      KAFKA_CLUSTERS_0_JMXPASSWORD: password
+      JAVA_OPTS: >-
+        -agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=*:5005
+        -Djavax.net.ssl.trustStore=/jmx/clienttruststore
+        -Djavax.net.ssl.trustStorePassword=12345678
+        -Djavax.net.ssl.keyStore=/jmx/clientkeystore
+        -Djavax.net.ssl.keyStorePassword=12345678
+    volumes:
+      - ./jmx/clienttruststore:/jmx/clienttruststore
+      - ./jmx/clientkeystore:/jmx/clientkeystore
+
+  zookeeper0:
+    image: confluentinc/cp-zookeeper:5.2.4
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+    ports:
+      - 2181:2181
+
+  kafka0:
+    image: confluentinc/cp-kafka:5.3.1
+    depends_on:
+      - zookeeper0
+    ports:
+      - 9092:9092
+      - 9997:9997
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper0:2181
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka0:29092,PLAINTEXT_HOST://localhost:9092
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      JMX_PORT: 9997
+      # CHMOD 700 FOR JMXREMOTE.* FILES
+      KAFKA_JMX_OPTS: >-
+        -Dcom.sun.management.jmxremote
+        -Dcom.sun.management.jmxremote.authenticate=true
+        -Dcom.sun.management.jmxremote.ssl=true
+        -Dcom.sun.management.jmxremote.registry.ssl=true
+        -Dcom.sun.management.jmxremote.ssl.need.client.auth=true
+        -Djavax.net.ssl.keyStore=/jmx/serverkeystore
+        -Djavax.net.ssl.keyStorePassword=12345678
+        -Djavax.net.ssl.trustStore=/jmx/servertruststore
+        -Djavax.net.ssl.trustStorePassword=12345678
+        -Dcom.sun.management.jmxremote.password.file=/jmx/jmxremote.password
+        -Dcom.sun.management.jmxremote.access.file=/jmx/jmxremote.access
+        -Dcom.sun.management.jmxremote.rmi.port=9997
+        -Djava.rmi.server.hostname=kafka0
+        -Djava.rmi.server.logCalls=true
+#        -Djavax.net.debug=ssl:handshake
+    volumes:
+    - ./jmx/serverkeystore:/jmx/serverkeystore
+    - ./jmx/servertruststore:/jmx/servertruststore
+    - ./jmx/jmxremote.password:/jmx/jmxremote.password
+    - ./jmx/jmxremote.access:/jmx/jmxremote.access
+
+  schemaregistry0:
+    image: confluentinc/cp-schema-registry:5.2.4
+    ports:
+      - 8085:8085
+    depends_on:
+      - zookeeper0
+      - kafka0
+    environment:
+      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: PLAINTEXT://kafka0:29092
+      SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: zookeeper0:2181
+      SCHEMA_REGISTRY_KAFKASTORE_SECURITY_PROTOCOL: PLAINTEXT
+      SCHEMA_REGISTRY_HOST_NAME: schemaregistry0
+      SCHEMA_REGISTRY_LISTENERS: http://schemaregistry0:8085
+
+      SCHEMA_REGISTRY_SCHEMA_REGISTRY_INTER_INSTANCE_PROTOCOL: "http"
+      SCHEMA_REGISTRY_LOG4J_ROOT_LOGLEVEL: INFO
+      SCHEMA_REGISTRY_KAFKASTORE_TOPIC: _schemas
+
+  kafka-connect0:
+    image: confluentinc/cp-kafka-connect:5.2.4
+    ports:
+      - 8083:8083
+    depends_on:
+      - kafka0
+      - schemaregistry0
+    environment:
+      CONNECT_BOOTSTRAP_SERVERS: kafka0:29092
+      CONNECT_GROUP_ID: compose-connect-group
+      CONNECT_CONFIG_STORAGE_TOPIC: _connect_configs
+      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
+      CONNECT_OFFSET_STORAGE_TOPIC: _connect_offset
+      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
+      CONNECT_STATUS_STORAGE_TOPIC: _connect_status
+      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
+      CONNECT_KEY_CONVERTER: org.apache.kafka.connect.storage.StringConverter
+      CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL: http://schemaregistry0:8085
+      CONNECT_VALUE_CONVERTER: org.apache.kafka.connect.storage.StringConverter
+      CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: http://schemaregistry0:8085
+      CONNECT_INTERNAL_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
+      CONNECT_INTERNAL_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
+      CONNECT_REST_ADVERTISED_HOST_NAME: kafka-connect0
+      CONNECT_PLUGIN_PATH: "/usr/share/java,/usr/share/confluent-hub-components"
+
+  kafka-init-topics:
+    image: confluentinc/cp-kafka:5.2.4
+    volumes:
+      - ./message.json:/data/message.json
+    depends_on:
+      - kafka0
+    command: "bash -c 'echo Waiting for Kafka to be ready... && \
+               cub kafka-ready -b kafka0:29092 1 30 && \
+               kafka-topics --create --topic second.users --partitions 3 --replication-factor 1 --if-not-exists --zookeeper zookeeper0:2181 && \
+               kafka-topics --create --topic first.messages --partitions 2 --replication-factor 1 --if-not-exists --zookeeper zookeeper0:2181 && \
+               kafka-console-producer --broker-list kafka0:29092 -topic second.users < /data/message.json'"

docker/kafka-ui-zookeeper-ssl.yml

@@ -7,7 +7,6 @@ services:
     image: provectuslabs/kafka-ui:latest
     ports:
       - 8080:8080
-      - 5005:5005
     volumes:
       - /tmp/kafka/secrets/kafka.kafka1.keystore.jks:/etc/kafka/secrets/kafka.zookeeper.keystore.jks
       - /tmp/kafka/secrets/kafka.zookeeper.truststore.jks:/etc/kafka/secrets/kafka.zookeeper.truststore.jks

kafka-ui-api/Dockerfile

@@ -3,5 +3,8 @@ VOLUME /tmp
 ARG JAR_FILE
 COPY "/target/${JAR_FILE}" "/kafka-ui-api.jar"
 
+ENV JAVA_OPTS=
+
 EXPOSE 8080
-CMD java -jar kafka-ui-api.jar
+
+CMD java $JAVA_OPTS -jar kafka-ui-api.jar

kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/ClustersProperties.java

@@ -28,6 +28,9 @@ public class ClustersProperties {
     String protobufMessageName;
     List<ConnectCluster> kafkaConnect;
     int jmxPort;
+    boolean jmxSsl;
+    String jmxUsername;
+    String jmxPassword;
     Properties properties;
     boolean readOnly = false;
     boolean disableLogDirsCollection = false;

kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/Config.java

@@ -1,5 +1,6 @@
 package com.provectus.kafka.ui.config;
 
+import com.provectus.kafka.ui.model.JmxConnectionInfo;
 import com.provectus.kafka.ui.util.JmxPoolFactory;
 import javax.management.remote.JMXConnector;
 import org.apache.commons.pool2.KeyedObjectPool;
@@ -14,9 +15,8 @@ import org.springframework.web.reactive.function.client.WebClient;
 public class Config {
 
   @Bean
-  public KeyedObjectPool<String, JMXConnector> pool() {
-    GenericKeyedObjectPool<String, JMXConnector> pool =
-        new GenericKeyedObjectPool<>(new JmxPoolFactory());
+  public KeyedObjectPool<JmxConnectionInfo, JMXConnector> pool() {
+    var pool = new GenericKeyedObjectPool<>(new JmxPoolFactory());
     pool.setConfig(poolConfig());
     return pool;
   }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/JmxConnectionInfo.java

@@ -0,0 +1,26 @@
+package com.provectus.kafka.ui.model;
+
+import lombok.Builder;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.RequiredArgsConstructor;
+
+@Data
+@RequiredArgsConstructor
+@Builder
+@EqualsAndHashCode(onlyExplicitlyIncluded = true)
+public class JmxConnectionInfo {
+
+  @EqualsAndHashCode.Include
+  private final String url;
+  private final boolean ssl;
+  private final String username;
+  private final String password;
+
+  public JmxConnectionInfo(String url) {
+    this.url = url;
+    this.ssl = false;
+    this.username = null;
+    this.password = null;
+  }
+}

kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/KafkaCluster.java

@@ -16,6 +16,9 @@ public class KafkaCluster {
   private final String name;
   private final String version;
   private final Integer jmxPort;
+  private final boolean jmxSsl;
+  private final String jmxUsername;
+  private final String jmxPassword;
   private final String bootstrapServers;
   private final String zookeeper;
   private final InternalSchemaRegistry schemaRegistry;

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaService.java

@@ -604,7 +604,8 @@ public class KafkaService {
     return clustersStorage.getClusterByName(clusterName)
         .filter(c -> c.getJmxPort() != null)
         .filter(c -> c.getJmxPort() > 0)
-        .map(c -> jmxClusterUtil.getJmxMetrics(c.getJmxPort(), node.host()))
+        .map(c -> jmxClusterUtil.getJmxMetrics(node.host(), c.getJmxPort(), c.isJmxSsl(),
+                c.getJmxUsername(), c.getJmxPassword()))
         .orElse(Collections.emptyList());
   }
 

kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/JmxClusterUtil.java

@@ -1,5 +1,6 @@
 package com.provectus.kafka.ui.util;
 
+import com.provectus.kafka.ui.model.JmxConnectionInfo;
 import com.provectus.kafka.ui.model.Metric;
 import java.math.BigDecimal;
 import java.util.ArrayList;
@@ -20,6 +21,7 @@ import lombok.RequiredArgsConstructor;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.pool2.KeyedObjectPool;
+import org.jetbrains.annotations.Nullable;
 import org.springframework.stereotype.Component;
 
 @Component
@@ -31,14 +33,21 @@ public class JmxClusterUtil {
   private static final String JMX_SERVICE_TYPE = "jmxrmi";
   private static final String KAFKA_SERVER_PARAM = "kafka.server";
   private static final String NAME_METRIC_FIELD = "name";
-  private final KeyedObjectPool<String, JMXConnector> pool;
+  private final KeyedObjectPool<JmxConnectionInfo, JMXConnector> pool;
 
   @SneakyThrows
-  public List<Metric> getJmxMetrics(int jmxPort, String jmxHost) {
-    String jmxUrl = JMX_URL + jmxHost + ":" + jmxPort + "/" + JMX_SERVICE_TYPE;
+  public List<Metric> getJmxMetrics(String host, int port, boolean jmxSsl,
+                                    @Nullable String username, @Nullable String password) {
+    String jmxUrl = JMX_URL + host + ":" + port + "/" + JMX_SERVICE_TYPE;
+    final var connectionInfo = JmxConnectionInfo.builder()
+            .url(jmxUrl)
+            .ssl(jmxSsl)
+            .username(username)
+            .password(password)
+            .build();
     JMXConnector srv;
     try {
-      srv = pool.borrowObject(jmxUrl);
+      srv = pool.borrowObject(connectionInfo);
     } catch (Exception e) {
       log.error("Cannot get JMX connector for the pool due to: ", e);
       return Collections.emptyList();
@@ -59,7 +68,7 @@ public class JmxClusterUtil {
         metric.setValue(getJmxMetric(jmxMetric.getCanonicalName(), msc));
         result.add(metric);
       }
-      pool.returnObject(jmxUrl, srv);
+      pool.returnObject(connectionInfo, srv);
     } catch (Exception e) {
       log.error("Cannot get jmxMetricsNames, {}", jmxUrl, e);
       closeConnectionExceptionally(jmxUrl, srv);
@@ -84,7 +93,7 @@ public class JmxClusterUtil {
 
   private void closeConnectionExceptionally(String url, JMXConnector srv) {
     try {
-      pool.invalidateObject(url, srv);
+      pool.invalidateObject(new JmxConnectionInfo(url), srv);
     } catch (Exception e) {
       log.error("Cannot invalidate object in pool, {}", url);
     }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/util/JmxPoolFactory.java

@@ -1,20 +1,34 @@
 package com.provectus.kafka.ui.util;
 
+import com.provectus.kafka.ui.model.JmxConnectionInfo;
 import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
 import javax.management.remote.JMXConnector;
 import javax.management.remote.JMXConnectorFactory;
 import javax.management.remote.JMXServiceURL;
+import javax.rmi.ssl.SslRMIClientSocketFactory;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.pool2.BaseKeyedPooledObjectFactory;
 import org.apache.commons.pool2.PooledObject;
 import org.apache.commons.pool2.impl.DefaultPooledObject;
 
 @Slf4j
-public class JmxPoolFactory extends BaseKeyedPooledObjectFactory<String, JMXConnector> {
+public class JmxPoolFactory extends BaseKeyedPooledObjectFactory<JmxConnectionInfo, JMXConnector> {
 
   @Override
-  public JMXConnector create(String s) throws Exception {
-    return JMXConnectorFactory.connect(new JMXServiceURL(s));
+  public JMXConnector create(JmxConnectionInfo info) throws Exception {
+    Map<String, Object> env = new HashMap<>();
+    if (StringUtils.isNotEmpty(info.getUsername()) && StringUtils.isNotEmpty(info.getPassword())) {
+      env.put("jmx.remote.credentials", new String[]{info.getUsername(), info.getPassword()});
+    }
+
+    if (info.isSsl()) {
+      env.put("com.sun.jndi.rmi.factory.socket", new SslRMIClientSocketFactory());
+    }
+
+    return JMXConnectorFactory.connect(new JMXServiceURL(info.getUrl()), env);
   }
 
   @Override
@@ -23,7 +37,7 @@ public class JmxPoolFactory extends BaseKeyedPooledObjectFactory<String, JMXConn
   }
 
   @Override
-  public void destroyObject(String key, PooledObject<JMXConnector> p) {
+  public void destroyObject(JmxConnectionInfo key, PooledObject<JMXConnector> p) {
     try {
       p.getObject().close();
     } catch (IOException e) {