Skip to content
This repository has been archived by the owner on Jan 24, 2024. It is now read-only.

[security] Communications between broker inherit BrokerClient configuration #1969

Merged
merged 1 commit into from
Jul 30, 2023
Merged

[security] Communications between broker inherit BrokerClient configuration #1969

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/ssl/SSLUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -302,9 +302,15 @@ public static SslContextFactory.Client createClientSslContextFactory(
break;
case SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG:
obj = kafkaServiceConfiguration.getKopSslTruststoreLocation();
if (obj == null) {
obj = kafkaServiceConfiguration.getBrokerClientTlsTrustStore();
}
break;
case SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG:
obj = kafkaServiceConfiguration.getKopSslTruststorePassword();
if (obj == null && kafkaServiceConfiguration.getKopSslTruststoreLocation() == null) {
obj = kafkaServiceConfiguration.getBrokerClientTlsTrustStorePassword();
}
break;
case SslConfigs.SSL_KEYMANAGER_ALGORITHM_CONFIG:
obj = kafkaServiceConfiguration.getKopSslKeymanagerAlgorithm();
Expand Down
23 changes: 16 additions & 7 deletions tests/src/test/java/io/streamnative/pulsar/handlers/kop/security/KafkaSSLChannelTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,7 @@ public class KafkaSSLChannelTest extends KopProtocolHandlerTestBase {
private String kopClientTruststoreLocation;
private String kopClientTruststorePassword;
private final boolean withCertHost;
private final boolean useBrokerClientTrustore;

static {
final HostnameVerifier defaultHostnameVerifier = javax.net.ssl.HttpsURLConnection.getDefaultHostnameVerifier();
Expand All @@ -73,9 +74,10 @@ public boolean verify(String hostname, javax.net.ssl.SSLSession sslSession) {
javax.net.ssl.HttpsURLConnection.setDefaultHostnameVerifier(localhostAcceptedHostnameVerifier);
}

public KafkaSSLChannelTest(final String entryFormat, boolean withCertHost) {
public KafkaSSLChannelTest(final String entryFormat, boolean withCertHost, boolean useBrokerClientTrustore) {
super(entryFormat);
this.withCertHost = withCertHost;
this.useBrokerClientTrustore = useBrokerClientTrustore;
setSslConfigurations(withCertHost);
}

Expand Down Expand Up @@ -103,10 +105,12 @@ private void setSslConfigurations(boolean withCertHost) {
@Factory
public static Object[] instances() {
return new Object[] {
new KafkaSSLChannelTest("pulsar", false),
new KafkaSSLChannelTest("pulsar", true),
new KafkaSSLChannelTest("kafka", false),
new KafkaSSLChannelTest("kafka", true)
new KafkaSSLChannelTest("pulsar", false, false),
new KafkaSSLChannelTest("pulsar", true, false),
new KafkaSSLChannelTest("kafka", false, false),
new KafkaSSLChannelTest("kafka", true, false),
// test rokerClientTlsTrustStore
new KafkaSSLChannelTest("kafka", true, true)
};
}

Expand All @@ -116,8 +120,13 @@ protected void sslSetUpForBroker() throws Exception {
conf.setKopSslKeystoreType("JKS");
conf.setKopSslKeystoreLocation(kopSslKeystoreLocation);
conf.setKopSslKeystorePassword(kopSslKeystorePassword);
conf.setKopSslTruststoreLocation(kopSslTruststoreLocation);
conf.setKopSslTruststorePassword(kopSslTruststorePassword);
if (useBrokerClientTrustore) {
conf.setBrokerClientTlsTrustStore(kopSslTruststoreLocation);
conf.setBrokerClientTlsTrustStorePassword(kopSslTruststorePassword);
} else {
conf.setKopSslTruststoreLocation(kopSslTruststoreLocation);
conf.setKopSslTruststorePassword(kopSslTruststorePassword);
}
}

@BeforeMethod
Expand Down
Loading