Skip to content

Commit

Permalink
[fix][client][branch-3.0] Fix compatibility between kerberos and tls (a…
Browse files Browse the repository at this point in the history 
…pache#23801)

Signed-off-by: Zixuan Liu <[email protected]>
(cherry picked from commit 4b69b30)
  • Loading branch information
@nodece @srinath-ctds
nodece authored and srinath-ctds committed Jan 3, 2025
1 parent b22764f commit 2f856db
Show file tree
Hide file tree
Showing 12 changed files with 227 additions and 140 deletions.
69 changes: 69 additions & 0 deletions pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsProducerConsumerTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,18 +18,24 @@
*/
package org.apache.pulsar.client.api;

import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Arrays;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;
import lombok.Cleanup;
import org.apache.commons.compress.utils.IOUtils;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.impl.auth.AuthenticationTls;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down Expand Up @@ -293,4 +299,67 @@ public void testTlsTransport(Supplier<String> url, Authentication auth) throws E
@Cleanup
Producer<byte[]> ignored = client.newProducer().topic(topicName).create();
}

@Test
public void testTlsWithFakeAuthentication() throws Exception {
Authentication authentication = spy(new Authentication() {
@Override
public String getAuthMethodName() {
return "fake";
}

@Override
public void configure(Map<String, String> authParams) {

}

@Override
public void start() {

}

@Override
public void close() {

}

@Override
public AuthenticationDataProvider getAuthData(String brokerHostName) {
return mock(AuthenticationDataProvider.class);
}
});

@Cleanup
PulsarAdmin pulsarAdmin = PulsarAdmin.builder()
.serviceHttpUrl(getPulsar().getWebServiceAddressTls())
.tlsTrustCertsFilePath(CA_CERT_FILE_PATH)
.allowTlsInsecureConnection(false)
.enableTlsHostnameVerification(false)
.tlsKeyFilePath(getTlsFileForClient("admin.key-pk8"))
.tlsCertificateFilePath(getTlsFileForClient("admin.cert"))
.authentication(authentication)
.build();
pulsarAdmin.tenants().getTenants();
verify(authentication, never()).getAuthData();

@Cleanup
PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(getPulsar().getBrokerServiceUrlTls())
.tlsTrustCertsFilePath(CA_CERT_FILE_PATH)
.allowTlsInsecureConnection(false)
.enableTlsHostnameVerification(false)
.tlsKeyFilePath(getTlsFileForClient("admin.key-pk8"))
.tlsCertificateFilePath(getTlsFileForClient("admin.cert"))
.authentication(authentication).build();
verify(authentication, never()).getAuthData();

final String topicName = "persistent://my-property/my-ns/my-topic-1";
internalSetUpForNamespace();
@Cleanup
Consumer<byte[]> ignoredConsumer =
pulsarClient.newConsumer().topic(topicName).subscriptionName("my-subscriber-name").subscribe();
verify(authentication, never()).getAuthData();
@Cleanup
Producer<byte[]> ignoredProducer = pulsarClient.newProducer().topic(topicName).create();
verify(authentication, never()).getAuthData();
}
}
3 changes: 2 additions & 1 deletion ...-admin/src/main/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnector.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -186,7 +186,8 @@ private void configureAsyncHttpClientSslEngineFactory(ClientConfigurationData co
DefaultAsyncHttpClientConfig.Builder confBuilder)
throws GeneralSecurityException, IOException {
// Set client key and certificate if available
AuthenticationDataProvider authData = conf.getAuthentication().getAuthData();
AuthenticationDataProvider authData = conf.getAuthentication().getAuthData(serviceNameResolver
.resolveHostUri().getHost());

SslEngineFactory sslEngineFactory = null;
if (conf.isUseKeyStoreTls()) {
Expand Down
1 change: 1 addition & 0 deletions pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Authentication.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,7 @@ public interface Authentication extends Closeable, Serializable {
* @throws PulsarClientException
* any other error
*/
@Deprecated
default AuthenticationDataProvider getAuthData() throws PulsarClientException {
throw new UnsupportedAuthenticationException("Method not implemented!");
}
Expand Down
2 changes: 1 addition & 1 deletion pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/CmdConsume.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -254,7 +254,7 @@ private int consumeFromWebSocket(String topic) {
try {
if (authentication != null) {
authentication.start();
AuthenticationDataProvider authData = authentication.getAuthData();
AuthenticationDataProvider authData = authentication.getAuthData(consumerUri.getHost());
if (authData.hasDataForHttp()) {
for (Map.Entry<String, String> kv : authData.getHttpHeaders()) {
consumeRequest.setHeader(kv.getKey(), kv.getValue());
Expand Down
2 changes: 1 addition & 1 deletion pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/CmdProduce.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -469,7 +469,7 @@ private int publishToWebSocket(String topic) {
try {
if (authentication != null) {
authentication.start();
AuthenticationDataProvider authData = authentication.getAuthData();
AuthenticationDataProvider authData = authentication.getAuthData(produceUri.getHost());
if (authData.hasDataForHttp()) {
for (Map.Entry<String, String> kv : authData.getHttpHeaders()) {
produceRequest.setHeader(kv.getKey(), kv.getValue());
Expand Down
2 changes: 1 addition & 1 deletion pulsar-client-tools/src/main/java/org/apache/pulsar/client/cli/CmdRead.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -243,7 +243,7 @@ private int readFromWebSocket(String topic) {
try {
if (authentication != null) {
authentication.start();
AuthenticationDataProvider authData = authentication.getAuthData();
AuthenticationDataProvider authData = authentication.getAuthData(readerUri.getHost());
if (authData.hasDataForHttp()) {
for (Map.Entry<String, String> kv : authData.getHttpHeaders()) {
readRequest.setHeader(kv.getKey(), kv.getValue());
Expand Down
3 changes: 2 additions & 1 deletion pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpClient.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,8 @@ public boolean keepAlive(InetSocketAddress remoteAddress, Request ahcRequest,
if ("https".equals(serviceNameResolver.getServiceUri().getServiceName())) {
try {
// Set client key and certificate if available
AuthenticationDataProvider authData = authentication.getAuthData();
AuthenticationDataProvider authData =
authentication.getAuthData(serviceNameResolver.resolveHostUri().getHost());

if (conf.isUseKeyStoreTls()) {
SSLContext sslCtx = null;
Expand Down
146 changes: 78 additions & 68 deletions pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarChannelInitializer.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,8 +29,10 @@
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.ssl.SslProvider;
import java.net.InetSocketAddress;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import lombok.Getter;
Expand Down Expand Up @@ -59,9 +61,9 @@ public class PulsarChannelInitializer extends ChannelInitializer<SocketChannel>
private final InetSocketAddress socks5ProxyAddress;
private final String socks5ProxyUsername;
private final String socks5ProxyPassword;

private final Supplier<SslContext> sslContextSupplier;
private NettySSLContextAutoRefreshBuilder nettySSLContextAutoRefreshBuilder;
private final ClientConfigurationData conf;
private Map<String, Supplier<SslContext>> sslContextSupplierMap;
private Map<String, NettySSLContextAutoRefreshBuilder> nettySSLContextAutoRefreshBuilderMap;

private static final long TLS_CERTIFICATE_CACHE_MILLIS = TimeUnit.MINUTES.toMillis(1);

Expand All @@ -76,15 +78,34 @@ public PulsarChannelInitializer(ClientConfigurationData conf, Supplier<ClientCnx
this.socks5ProxyPassword = conf.getSocks5ProxyPassword();

this.tlsEnabledWithKeyStore = conf.isUseKeyStoreTls();
this.conf = conf.clone();
this.sslContextSupplierMap = new ConcurrentHashMap<>();
this.nettySSLContextAutoRefreshBuilderMap = new ConcurrentHashMap<>();
}

if (tlsEnabled) {
if (tlsEnabledWithKeyStore) {
AuthenticationDataProvider authData1 = conf.getAuthentication().getAuthData();
if (StringUtils.isBlank(conf.getTlsTrustStorePath())) {
throw new PulsarClientException("Failed to create TLS context, the tlsTrustStorePath"
+ " need to be configured if useKeyStoreTls enabled");
}
nettySSLContextAutoRefreshBuilder = new NettySSLContextAutoRefreshBuilder(
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast("consolidation", new FlushConsolidationHandler(1024, true));

// Setup channel except for the SsHandler for TLS enabled connections
ch.pipeline().addLast("ByteBufPairEncoder", ByteBufPair.getEncoder(tlsEnabled));

ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(
Commands.DEFAULT_MAX_MESSAGE_SIZE + Commands.MESSAGE_SIZE_FRAME_PADDING, 0, 4, 0, 4));
ChannelHandler clientCnx = clientCnxSupplier.get();
ch.pipeline().addLast("handler", clientCnx);
}

private NettySSLContextAutoRefreshBuilder getNettySSLContextAutoRefreshBuilder(String host)
throws PulsarClientException {
if (tlsEnabledWithKeyStore) {
AuthenticationDataProvider authData1 = conf.getAuthentication().getAuthData(host);
if (StringUtils.isBlank(conf.getTlsTrustStorePath())) {
throw new PulsarClientException("Failed to create TLS context, the tlsTrustStorePath"
+ " need to be configured if useKeyStoreTls enabled");
}
return nettySSLContextAutoRefreshBuilderMap.computeIfAbsent(host,
key -> new NettySSLContextAutoRefreshBuilder(
conf.getSslProvider(),
conf.isTlsAllowInsecureConnection(),
conf.getTlsTrustStoreType(),
Expand All @@ -96,64 +117,52 @@ public PulsarChannelInitializer(ClientConfigurationData conf, Supplier<ClientCnx
conf.getTlsCiphers(),
conf.getTlsProtocols(),
TLS_CERTIFICATE_CACHE_MILLIS,
authData1);
}

sslContextSupplier = new ObjectCache<SslContext>(() -> {
try {
SslProvider sslProvider = null;
if (conf.getSslProvider() != null) {
sslProvider = SslProvider.valueOf(conf.getSslProvider());
}

// Set client certificate if available
AuthenticationDataProvider authData = conf.getAuthentication().getAuthData();
if (authData.hasDataForTls()) {
return authData.getTlsTrustStoreStream() == null
? SecurityUtility.createNettySslContextForClient(
sslProvider,
conf.isTlsAllowInsecureConnection(),
conf.getTlsTrustCertsFilePath(),
authData.getTlsCertificates(),
authData.getTlsPrivateKey(),
conf.getTlsCiphers(),
conf.getTlsProtocols())
: SecurityUtility.createNettySslContextForClient(sslProvider,
conf.isTlsAllowInsecureConnection(),
authData.getTlsTrustStoreStream(),
authData.getTlsCertificates(), authData.getTlsPrivateKey(),
conf.getTlsCiphers(),
conf.getTlsProtocols());
} else {
return SecurityUtility.createNettySslContextForClient(
sslProvider,
conf.isTlsAllowInsecureConnection(),
conf.getTlsTrustCertsFilePath(),
conf.getTlsCertificateFilePath(),
conf.getTlsKeyFilePath(),
conf.getTlsCiphers(),
conf.getTlsProtocols());
}
} catch (Exception e) {
throw new RuntimeException("Failed to create TLS context", e);
}
}, TLS_CERTIFICATE_CACHE_MILLIS, TimeUnit.MILLISECONDS);
} else {
sslContextSupplier = null;
authData1));
}
throw new PulsarClientException(
"Failed to create TLS context, the tlsEnabledWithKeyStore need to be true");
}

@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast("consolidation", new FlushConsolidationHandler(1024, true));

// Setup channel except for the SsHandler for TLS enabled connections
ch.pipeline().addLast("ByteBufPairEncoder", ByteBufPair.getEncoder(tlsEnabled));
private Supplier<SslContext> getSslContextSupplier(String host) {
return sslContextSupplierMap.computeIfAbsent(host, key -> new ObjectCache<>(() -> {
try {
SslProvider sslProvider = null;
if (conf.getSslProvider() != null) {
sslProvider = SslProvider.valueOf(conf.getSslProvider());
}

ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(
Commands.DEFAULT_MAX_MESSAGE_SIZE + Commands.MESSAGE_SIZE_FRAME_PADDING, 0, 4, 0, 4));
ChannelHandler clientCnx = clientCnxSupplier.get();
ch.pipeline().addLast("handler", clientCnx);
// Set client certificate if available
AuthenticationDataProvider authData = conf.getAuthentication().getAuthData(host);
if (authData.hasDataForTls()) {
return authData.getTlsTrustStoreStream() == null
? SecurityUtility.createNettySslContextForClient(
sslProvider,
conf.isTlsAllowInsecureConnection(),
conf.getTlsTrustCertsFilePath(),
authData.getTlsCertificates(),
authData.getTlsPrivateKey(),
conf.getTlsCiphers(),
conf.getTlsProtocols())
: SecurityUtility.createNettySslContextForClient(sslProvider,
conf.isTlsAllowInsecureConnection(),
authData.getTlsTrustStoreStream(),
authData.getTlsCertificates(), authData.getTlsPrivateKey(),
conf.getTlsCiphers(),
conf.getTlsProtocols());
} else {
return SecurityUtility.createNettySslContextForClient(
sslProvider,
conf.isTlsAllowInsecureConnection(),
conf.getTlsTrustCertsFilePath(),
conf.getTlsCertificateFilePath(),
conf.getTlsKeyFilePath(),
conf.getTlsCiphers(),
conf.getTlsProtocols());
}
} catch (Exception e) {
throw new RuntimeException("Failed to create TLS context", e);
}
}, TLS_CERTIFICATE_CACHE_MILLIS, TimeUnit.MILLISECONDS));
}

/**
Expand All @@ -175,9 +184,10 @@ CompletableFuture<Channel> initTls(Channel ch, InetSocketAddress sniHost) {
ch.eventLoop().execute(() -> {
try {
SslHandler handler = tlsEnabledWithKeyStore
? new SslHandler(nettySSLContextAutoRefreshBuilder.get()
.createSSLEngine(sniHost.getHostString(), sniHost.getPort()))
: sslContextSupplier.get().newHandler(ch.alloc(), sniHost.getHostString(), sniHost.getPort());
? new SslHandler(getNettySSLContextAutoRefreshBuilder(sniHost.getHostName()).get()
.createSSLEngine(sniHost.getHostString(), sniHost.getPort()))
: getSslContextSupplier(sniHost.getHostName()).get()
.newHandler(ch.alloc(), sniHost.getHostString(), sniHost.getPort());

if (tlsHostnameVerificationEnabled) {
SecurityUtility.configureSSLHandler(handler);
Expand Down
Loading

0 comments on commit 2f856db

Please sign in to comment.