Skip to content

Commit 858f01d

Browse files
CrimCrim
authored
Add support for connecting to SASL authenticated Kafka clusters. (#109)
* Proof of Concept for SASL plaintext auth * Start refactoring common Kafka client configuration into separate utility class * Start updating cluster create/update UI to support SASL options * drop dead fields * Use flyway to migrate database tables * Update Cluster model for new fields * Wire in UI to controller to persist cluster with SASL settings * Add test coverage to Cluster controller for setting up sasl clusters * fix bad assertions * Wire in SASL options from Cluster * Add test coverage * Rearrange tests * add missing headers * WIP - add customized listeners? * Enable SSL, SASL support in DevCluster * Add test coverage for SSL, SASL, and SASL_SSL clusters * update gitignore * fix travis * fix travis * fix travis * fix travis * Fix failing test * add missing imports
1 parent aea710f commit 858f01d

40 files changed

+2515
-159
lines changed

‎.gitignore‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,4 +5,7 @@ kafka-webview-ui/src/main/frontend/node_modules/
55
kafka-webview-ui/src/main/frontend/dist/
66
kafka-webview-ui/src/main/resources/static/
77
data/
8-
TODO
8+
kafka-webview-ui/src/test/resources/kafka.keystore.jks
9+
kafka-webview-ui/src/test/resources/kafka.truststore.jks
10+
dev-cluster/src/main/resources/kafka.keystore.jks
11+
dev-cluster/src/main/resources/kafka.truststore.jks

‎.travis.yml‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,5 +6,12 @@ cache:
66
directories:
77
- $HOME/.m2
88
- kafka-webview-ui/src/main/frontend/node_modules/
9-
before_script:
10-
- mkdir -p ./data/uploads
9+
script:
10+
## Ensure data directory exists
11+
- mkdir -p ./data/uploads
12+
## Generate dummy SSL certificates
13+
- ./generateDummySslCertificates.sh
14+
## Run CheckStyle and License Header checks, compile, and install locally
15+
- mvn clean install -DskipTests=true -DskipCheckStyle=false -Dmaven.javadoc.skip=true -B -V
16+
## Run Test Suite
17+
- mvn test -B -DskipCheckStyle=true -Djava.security.auth.login.config=${PWD}/kafka-webview-ui/src/test/resources/jaas.conf

‎CHANGELOG.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,9 +4,11 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
44

55
## 2.1.0 (UNRELEASED)
66
#### New Features
7+
- Add support for SASL authentication to Kafka clusters.
78
- Consumer details can now be viewed on Clusters.
89
- Explicitly expose user login session timeout configuration value in documentation.
910

11+
1012
#### Bug fixes
1113
- Topics shown on brokers now include "internal" topics.
1214
- Generated consumer client.id shortened.

‎dev-cluster/pom.xml‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,7 @@
5252
<dependency>
5353
<groupId>com.salesforce.kafka.test</groupId>
5454
<artifactId>kafka-junit-core</artifactId>
55-
<version>3.0.1</version>
55+
<version>3.1.0</version>
5656
</dependency>
5757

5858
<!-- Include Kafka 1.1.x -->
@@ -67,6 +67,13 @@
6767
<version>1.1.1</version>
6868
</dependency>
6969

70+
<!-- Command line argument parsing -->
71+
<dependency>
72+
<groupId>commons-cli</groupId>
73+
<artifactId>commons-cli</artifactId>
74+
<version>1.4</version>
75+
</dependency>
76+
7077
<!-- LDAP Server -->
7178
<dependency>
7279
<groupId>com.unboundid</groupId>

‎dev-cluster/src/main/java/org/sourcelab/kafka/devcluster/DevCluster.java‎

Lines changed: 129 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -26,12 +26,25 @@
2626

2727
import com.salesforce.kafka.test.KafkaTestCluster;
2828
import com.salesforce.kafka.test.KafkaTestUtils;
29+
import com.salesforce.kafka.test.listeners.BrokerListener;
30+
import com.salesforce.kafka.test.listeners.PlainListener;
31+
import com.salesforce.kafka.test.listeners.SaslPlainListener;
32+
import com.salesforce.kafka.test.listeners.SaslSslListener;
33+
import com.salesforce.kafka.test.listeners.SslListener;
34+
import org.apache.commons.cli.CommandLine;
35+
import org.apache.commons.cli.CommandLineParser;
36+
import org.apache.commons.cli.DefaultParser;
37+
import org.apache.commons.cli.HelpFormatter;
38+
import org.apache.commons.cli.Option;
39+
import org.apache.commons.cli.Options;
40+
import org.apache.commons.cli.ParseException;
2941
import org.apache.kafka.clients.consumer.ConsumerRecords;
3042
import org.apache.kafka.clients.consumer.KafkaConsumer;
3143
import org.apache.kafka.common.serialization.StringDeserializer;
3244
import org.slf4j.Logger;
3345
import org.slf4j.LoggerFactory;
3446

47+
import java.net.URL;
3548
import java.util.Collections;
3649
import java.util.Properties;
3750

@@ -49,49 +62,147 @@ public class DevCluster {
4962
* @param args command line args.
5063
*/
5164
public static void main(final String[] args) throws Exception {
52-
// Right now we accept one parameter, the number of nodes in the cluster.
53-
final int clusterSize;
54-
if (args.length > 0) {
55-
clusterSize = Integer.parseInt(args[0]);
56-
} else {
57-
clusterSize = 1;
58-
}
65+
// Parse command line arguments
66+
final CommandLine cmd = parseArguments(args);
5967

68+
// Right now we accept one parameter, the number of nodes in the cluster.
69+
final int clusterSize = Integer.parseInt(cmd.getOptionValue("size"));
6070
logger.info("Starting up kafka cluster with {} brokers", clusterSize);
6171

72+
// Default to plaintext listener.
73+
BrokerListener listener = new PlainListener();
74+
75+
final URL trustStore = DevCluster.class.getClassLoader().getResource("kafka.truststore.jks");
76+
final URL keyStore = DevCluster.class.getClassLoader().getResource("kafka.keystore.jks");
77+
78+
final Properties properties = new Properties();
79+
if (cmd.hasOption("sasl") && cmd.hasOption("ssl")) {
80+
listener = new SaslSslListener()
81+
// SSL Options
82+
.withClientAuthRequired()
83+
.withTrustStoreLocation(trustStore.getFile())
84+
.withTrustStorePassword("password")
85+
.withKeyStoreLocation(keyStore.getFile())
86+
.withKeyStorePassword("password")
87+
.withKeyPassword("password")
88+
// SASL Options.
89+
.withUsername("kafkaclient")
90+
.withPassword("client-secret");
91+
} else if (cmd.hasOption("sasl")) {
92+
listener = new SaslPlainListener()
93+
.withUsername("kafkaclient")
94+
.withPassword("client-secret");
95+
} else if (cmd.hasOption("ssl")) {
96+
listener = new SslListener()
97+
.withClientAuthRequired()
98+
.withTrustStoreLocation(trustStore.getFile())
99+
.withTrustStorePassword("password")
100+
.withKeyStoreLocation(keyStore.getFile())
101+
.withKeyStorePassword("password")
102+
.withKeyPassword("password");
103+
}
104+
62105
// Create a test cluster
63-
final KafkaTestCluster kafkaTestCluster = new KafkaTestCluster(clusterSize);
106+
final KafkaTestCluster kafkaTestCluster = new KafkaTestCluster(
107+
clusterSize,
108+
properties,
109+
Collections.singletonList(listener)
110+
);
64111

65112
// Start the cluster.
66113
kafkaTestCluster.start();
67114

68-
// Create a topic
69-
final String topicName = "TestTopicA";
70-
final int partitionsCount = 5 * clusterSize;
71-
final KafkaTestUtils utils = new KafkaTestUtils(kafkaTestCluster);
72-
utils.createTopic(topicName, partitionsCount, (short) clusterSize);
115+
// Create topics
116+
String[] topicNames = null;
117+
if (cmd.hasOption("topic")) {
118+
topicNames = cmd.getOptionValues("topic");
119+
120+
for (final String topicName : topicNames) {
121+
final KafkaTestUtils utils = new KafkaTestUtils(kafkaTestCluster);
122+
utils.createTopic(topicName, clusterSize, (short) clusterSize);
73123

74-
// Publish some data into that topic
75-
for (int partition = 0; partition < partitionsCount; partition++) {
76-
utils.produceRecords(1000, topicName, partition);
124+
// Publish some data into that topic
125+
for (int partition = 0; partition < clusterSize; partition++) {
126+
utils.produceRecords(1000, topicName, partition);
127+
}
128+
}
77129
}
78130

131+
// Log topic names created.
132+
if (topicNames != null) {
133+
logger.info("Created topics: {}", String.join(", ", topicNames));
134+
}
135+
136+
// Log how to connect to cluster brokers.
79137
kafkaTestCluster
80138
.getKafkaBrokers()
81139
.stream()
82140
.forEach((broker) -> {
83141
logger.info("Started broker with Id {} at {}", broker.getBrokerId(), broker.getConnectString());
84142
});
85143

144+
// Log cluster connect string.
86145
logger.info("Cluster started at: {}", kafkaTestCluster.getKafkaConnectString());
87146

88-
runEndlessConsumer(topicName, utils);
89-
runEndlessProducer(topicName, partitionsCount, utils);
147+
//runEndlessConsumer(topicName, utils);
148+
//runEndlessProducer(topicName, partitionsCount, utils);
90149

91150
// Wait forever.
92151
Thread.currentThread().join();
93152
}
94153

154+
private static CommandLine parseArguments(final String[] args) throws ParseException {
155+
// create Options object
156+
final Options options = new Options();
157+
158+
// add number of brokers
159+
options.addOption(Option.builder("size")
160+
.desc("Number of brokers to start")
161+
.required()
162+
.hasArg()
163+
.type(Integer.class)
164+
.build()
165+
);
166+
167+
options.addOption(Option.builder("topic")
168+
.desc("Create test topic")
169+
.required(false)
170+
.hasArgs()
171+
.build()
172+
);
173+
174+
// Optionally enable SASL
175+
options.addOption(Option.builder("sasl")
176+
.desc("Enable SASL authentication")
177+
.required(false)
178+
.hasArg(false)
179+
.type(Boolean.class)
180+
.build()
181+
);
182+
183+
// Optionally enable SSL
184+
options.addOption(Option.builder("ssl")
185+
.desc("Enable SSL")
186+
.required(false)
187+
.hasArg(false)
188+
.type(Boolean.class)
189+
.build()
190+
);
191+
192+
try {
193+
final CommandLineParser parser = new DefaultParser();
194+
return parser.parse(options, args);
195+
} catch (final Exception exception) {
196+
System.out.println("ERROR: " + exception.getMessage() + "\n");
197+
final HelpFormatter formatter = new HelpFormatter();
198+
formatter.printHelp( "DevCluster", options);
199+
System.out.println("");
200+
System.out.flush();
201+
202+
throw exception;
203+
}
204+
}
205+
95206
/**
96207
* Fire up a new thread running an endless producer script into the given topic and partitions.
97208
* @param topicName Name of the topic to produce records into.

‎dev-cluster/src/main/resources/jaas.conf‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
KafkaServer {
2+
org.apache.kafka.common.security.plain.PlainLoginModule required
3+
username="admin"
4+
password="admin-secret"
5+
user_admin="admin-secret"
6+
user_kafkaclient="client-secret";
7+
};
8+
9+
Server {
10+
org.apache.zookeeper.server.auth.DigestLoginModule required
11+
username="admin"
12+
password="admin-secret"
13+
user_zooclient="client-secret";
14+
};
15+
16+
Client {
17+
org.apache.zookeeper.server.auth.DigestLoginModule required
18+
username="zooclient"
19+
password="client-secret";
20+
};

‎dev-cluster/src/main/resources/log4j2.xml‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,6 @@
1010
<Loggers>
1111
<Root level="info">
1212
<AppenderRef ref="STDOUT"/>
13-
<AppenderRef ref="FILE"/>
1413
</Root>
1514
</Loggers>
1615

‎generateDummySslCertificates.sh‎

Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,72 @@
1+
### This script generates a dummy self signed certificate used in the test suite.
2+
3+
#!/bin/bash
4+
cd "$(dirname "$0")"
5+
6+
set -e
7+
8+
KEYSTORE_FILENAME="kafka.keystore.jks"
9+
VALIDITY_IN_DAYS=3650
10+
DEFAULT_TRUSTSTORE_FILENAME="kafka.truststore.jks"
11+
TRUSTSTORE_WORKING_DIRECTORY="generated/truststore"
12+
KEYSTORE_WORKING_DIRECTORY="generated/keystore"
13+
CA_CERT_FILE="ca-cert"
14+
KEYSTORE_SIGN_REQUEST="cert-file"
15+
KEYSTORE_SIGN_REQUEST_SRL="ca-cert.srl"
16+
KEYSTORE_SIGNED_CERT="cert-signed"
17+
WEBVIEW_UI_DEST_DIRECTORY="kafka-webview-ui/src/test/resources/"
18+
DEV_CLUSTER_DEST_DIRECTORY="dev-cluster/src/main/resources/"
19+
20+
rm -rf generated
21+
mkdir -p $TRUSTSTORE_WORKING_DIRECTORY
22+
23+
openssl req -new -x509 -keyout $TRUSTSTORE_WORKING_DIRECTORY/ca-key \
24+
-out $TRUSTSTORE_WORKING_DIRECTORY/ca-cert -days $VALIDITY_IN_DAYS \
25+
-subj "/C=JP/ST=Tokyo/L=Japan/O=KafkaWebView/OU=Engineer/CN=localhost" \
26+
-passout pass:password
27+
28+
trust_store_private_key_file="$TRUSTSTORE_WORKING_DIRECTORY/ca-key"
29+
30+
keytool -keystore $TRUSTSTORE_WORKING_DIRECTORY/$DEFAULT_TRUSTSTORE_FILENAME \
31+
-alias CARoot -import -file $TRUSTSTORE_WORKING_DIRECTORY/ca-cert \
32+
-storepass password -trustcacerts -noprompt
33+
trust_store_file="$TRUSTSTORE_WORKING_DIRECTORY/$DEFAULT_TRUSTSTORE_FILENAME"
34+
35+
rm $TRUSTSTORE_WORKING_DIRECTORY/$CA_CERT_FILE
36+
37+
mkdir $KEYSTORE_WORKING_DIRECTORY
38+
keytool -keystore $KEYSTORE_WORKING_DIRECTORY/$KEYSTORE_FILENAME \
39+
-alias localhost -validity $VALIDITY_IN_DAYS -genkey -keyalg RSA \
40+
-storepass password -keypass password \
41+
-dname "CN=localhost, OU=localhost, O=localhost, L=localhost, ST=localhost, C=localhost"
42+
43+
keytool -keystore $trust_store_file -export -alias CARoot -rfc -file $CA_CERT_FILE \
44+
-storepass password -keypass password
45+
46+
keytool -keystore $KEYSTORE_WORKING_DIRECTORY/$KEYSTORE_FILENAME -alias localhost \
47+
-certreq -file $KEYSTORE_SIGN_REQUEST \
48+
-keypass password -storepass password
49+
50+
openssl x509 -req -CA $CA_CERT_FILE -CAkey $trust_store_private_key_file \
51+
-in $KEYSTORE_SIGN_REQUEST -out $KEYSTORE_SIGNED_CERT \
52+
-days $VALIDITY_IN_DAYS -CAcreateserial -passin pass:password
53+
54+
keytool -keystore $KEYSTORE_WORKING_DIRECTORY/$KEYSTORE_FILENAME -alias CARoot \
55+
-import -file $CA_CERT_FILE \
56+
-keypass password -storepass password -noprompt
57+
rm $CA_CERT_FILE
58+
59+
keytool -keystore $KEYSTORE_WORKING_DIRECTORY/$KEYSTORE_FILENAME -alias localhost -import \
60+
-file $KEYSTORE_SIGNED_CERT -storepass password -keypass password
61+
62+
rm $KEYSTORE_SIGN_REQUEST_SRL
63+
rm $KEYSTORE_SIGN_REQUEST
64+
rm $KEYSTORE_SIGNED_CERT
65+
rm $trust_store_private_key_file
66+
67+
cp -rp $KEYSTORE_WORKING_DIRECTORY/$KEYSTORE_FILENAME $WEBVIEW_UI_DEST_DIRECTORY
68+
cp -rp $KEYSTORE_WORKING_DIRECTORY/$KEYSTORE_FILENAME $DEV_CLUSTER_DEST_DIRECTORY
69+
cp -rp $trust_store_file $WEBVIEW_UI_DEST_DIRECTORY
70+
cp -rp $trust_store_file $DEV_CLUSTER_DEST_DIRECTORY
71+
72+
rm -rf generated

‎kafka-webview-ui/pom.xml‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -88,6 +88,12 @@
8888
<artifactId>h2</artifactId>
8989
</dependency>
9090

91+
<!-- Flyway for database migrations -->
92+
<dependency>
93+
<groupId>org.flywaydb</groupId>
94+
<artifactId>flyway-core</artifactId>
95+
</dependency>
96+
9197
<!-- Security Module -->
9298
<dependency>
9399
<groupId>org.springframework.boot</groupId>
@@ -182,7 +188,7 @@
182188
<dependency>
183189
<groupId>com.salesforce.kafka.test</groupId>
184190
<artifactId>kafka-junit4</artifactId>
185-
<version>3.0.1</version>
191+
<version>3.1.0</version>
186192
<scope>test</scope>
187193
</dependency>
188194
<dependency>

0 commit comments

Comments
 (0)