Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[fix][client][branch-3.0] Fix compatibility between kerberos and tls #23801

Merged
merged 1 commit into from
Jan 3, 2025
Merged

[fix][client][branch-3.0] Fix compatibility between kerberos and tls #23801

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
69 changes: 69 additions & 0 deletions pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsProducerConsumerTest.java
View file
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -401,7 +401,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
Open in desktop
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
Open in desktop
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
Open in desktop
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
Loading