Skip to content

Commit

Permalink
[fix][client] Fix compatibility between kerberos and tls (#23798)
Browse files Browse the repository at this point in the history 
Signed-off-by: Zixuan Liu <[email protected]>
  • Loading branch information
@nodece
nodece authored Jan 2, 2025
1 parent 4a01423 commit fd45029
Show file tree
Hide file tree
Showing 13 changed files with 182 additions and 80 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 @@ -295,4 +301,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();
}
}
7 changes: 4 additions & 3 deletions ...-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 @@ -191,7 +191,8 @@ private void configureAsyncHttpClientSslEngineFactory(ClientConfigurationData co
// Set client key and certificate if available
sslRefresher = Executors.newScheduledThreadPool(1,
new DefaultThreadFactory("pulsar-admin-ssl-refresher"));
PulsarSslConfiguration sslConfiguration = buildSslConfiguration(conf);
PulsarSslConfiguration sslConfiguration = buildSslConfiguration(conf, serviceNameResolver
.resolveHostUri().getHost());
this.sslFactory = (PulsarSslFactory) Class.forName(conf.getSslFactoryPlugin())
.getConstructor().newInstance();
this.sslFactory.initialize(sslConfiguration);
Expand Down Expand Up @@ -519,7 +520,7 @@ public void close() {
}
}

protected PulsarSslConfiguration buildSslConfiguration(ClientConfigurationData conf)
protected PulsarSslConfiguration buildSslConfiguration(ClientConfigurationData conf, String host)
throws PulsarClientException {
return PulsarSslConfiguration.builder()
.tlsProvider(conf.getSslProvider())
Expand All @@ -537,7 +538,7 @@ protected PulsarSslConfiguration buildSslConfiguration(ClientConfigurationData c
.allowInsecureConnection(conf.isTlsAllowInsecureConnection())
.requireTrustedClientCertOnConnect(false)
.tlsEnabledWithKeystore(conf.isUseKeyStoreTls())
.authData(conf.getAuthentication().getAuthData())
.authData(conf.getAuthentication().getAuthData(host))
.tlsCustomParams(conf.getSslFactoryPluginParams())
.serverMode(false)
.isHttps(true)
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 @@ -260,7 +260,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 @@ -244,7 +244,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
7 changes: 4 additions & 3 deletions 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 @@ -97,7 +97,8 @@ public boolean keepAlive(InetSocketAddress remoteAddress, Request ahcRequest,
this.executorService = Executors
.newSingleThreadScheduledExecutor(new ExecutorProvider
.ExtendedThreadFactory("httpclient-ssl-refresh"));
PulsarSslConfiguration sslConfiguration = buildSslConfiguration(conf);
PulsarSslConfiguration sslConfiguration =
buildSslConfiguration(conf, serviceNameResolver.resolveHostUri().getHost());
this.sslFactory = (PulsarSslFactory) Class.forName(conf.getSslFactoryPlugin())
.getConstructor().newInstance();
this.sslFactory.initialize(sslConfiguration);
Expand Down Expand Up @@ -233,7 +234,7 @@ public <T> CompletableFuture<T> get(String path, Class<T> clazz) {
return future;
}

protected PulsarSslConfiguration buildSslConfiguration(ClientConfigurationData config)
protected PulsarSslConfiguration buildSslConfiguration(ClientConfigurationData config, String host)
throws PulsarClientException {
return PulsarSslConfiguration.builder()
.tlsProvider(config.getSslProvider())
Expand All @@ -252,7 +253,7 @@ protected PulsarSslConfiguration buildSslConfiguration(ClientConfigurationData c
.requireTrustedClientCertOnConnect(false)
.tlsEnabledWithKeystore(config.isUseKeyStoreTls())
.tlsCustomParams(config.getSslFactoryPluginParams())
.authData(config.getAuthentication().getAuthData())
.authData(config.getAuthentication().getAuthData(host))
.serverMode(false)
.isHttps(true)
.build();
Expand Down
70 changes: 42 additions & 28 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 @@ -27,8 +27,10 @@
import io.netty.handler.proxy.Socks5ProxyHandler;
import io.netty.handler.ssl.SslHandler;
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.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
Expand All @@ -55,8 +57,8 @@ public class PulsarChannelInitializer extends ChannelInitializer<SocketChannel>
private final InetSocketAddress socks5ProxyAddress;
private final String socks5ProxyUsername;
private final String socks5ProxyPassword;

private final PulsarSslFactory pulsarSslFactory;
private final ClientConfigurationData conf;
private final Map<String, PulsarSslFactory> pulsarSslFactoryMap;

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

Expand All @@ -69,26 +71,17 @@ public PulsarChannelInitializer(ClientConfigurationData conf, Supplier<ClientCnx
this.socks5ProxyAddress = conf.getSocks5ProxyAddress();
this.socks5ProxyUsername = conf.getSocks5ProxyUsername();
this.socks5ProxyPassword = conf.getSocks5ProxyPassword();

this.conf = conf.clone();
if (tlsEnabled) {
this.pulsarSslFactory = (PulsarSslFactory) Class.forName(conf.getSslFactoryPlugin())
.getConstructor().newInstance();
try {
PulsarSslConfiguration sslConfiguration = buildSslConfiguration(conf);
this.pulsarSslFactory.initialize(sslConfiguration);
this.pulsarSslFactory.createInternalSslContext();
} catch (Exception e) {
log.error("Unable to initialize and create the ssl context", e);
}
this.pulsarSslFactoryMap = new ConcurrentHashMap<>();
if (scheduledExecutorService != null && conf.getAutoCertRefreshSeconds() > 0) {
scheduledExecutorService.scheduleWithFixedDelay(() -> this.refreshSslContext(conf),
conf.getAutoCertRefreshSeconds(),
conf.getAutoCertRefreshSeconds(),
TimeUnit.SECONDS);
}

} else {
pulsarSslFactory = null;
this.pulsarSslFactoryMap = null;
}
}

Expand Down Expand Up @@ -123,6 +116,23 @@ CompletableFuture<Channel> initTls(Channel ch, InetSocketAddress sniHost) {
CompletableFuture<Channel> initTlsFuture = new CompletableFuture<>();
ch.eventLoop().execute(() -> {
try {
PulsarSslFactory pulsarSslFactory = pulsarSslFactoryMap.computeIfAbsent(sniHost.getHostName(), key -> {
try {
PulsarSslFactory factory = (PulsarSslFactory) Class.forName(conf.getSslFactoryPlugin())
.getConstructor().newInstance();
PulsarSslConfiguration sslConfiguration = buildSslConfiguration(conf, key);
factory.initialize(sslConfiguration);
factory.createInternalSslContext();
return factory;
} catch (Exception e) {
log.error("Unable to initialize and create the ssl context", e);
initTlsFuture.completeExceptionally(e);
return null;
}
});
if (pulsarSslFactory == null) {
return;
}
SslHandler handler = new SslHandler(pulsarSslFactory
.createClientSslEngine(ch.alloc(), sniHost.getHostName(), sniHost.getPort()));

Expand Down Expand Up @@ -181,7 +191,9 @@ CompletableFuture<Channel> initializeClientCnx(Channel ch,
return ch;
}));
}
protected PulsarSslConfiguration buildSslConfiguration(ClientConfigurationData config)

protected PulsarSslConfiguration buildSslConfiguration(ClientConfigurationData config,
String host)
throws PulsarClientException {
return PulsarSslConfiguration.builder()
.tlsProvider(config.getSslProvider())
Expand All @@ -200,28 +212,30 @@ protected PulsarSslConfiguration buildSslConfiguration(ClientConfigurationData c
.requireTrustedClientCertOnConnect(false)
.tlsEnabledWithKeystore(config.isUseKeyStoreTls())
.tlsCustomParams(config.getSslFactoryPluginParams())
.authData(config.getAuthentication().getAuthData())
.authData(config.getAuthentication().getAuthData(host))
.serverMode(false)
.build();
}

protected void refreshSslContext(ClientConfigurationData conf) {
try {
pulsarSslFactoryMap.forEach((key, pulsarSslFactory) -> {
try {
if (conf.isUseKeyStoreTls()) {
this.pulsarSslFactory.getInternalSslContext();
} else {
this.pulsarSslFactory.getInternalNettySslContext();
try {
if (conf.isUseKeyStoreTls()) {
pulsarSslFactory.getInternalSslContext();
} else {
pulsarSslFactory.getInternalNettySslContext();
}
} catch (Exception e) {
log.error("SSL Context is not initialized", e);
PulsarSslConfiguration sslConfiguration = buildSslConfiguration(conf, key);
pulsarSslFactory.initialize(sslConfiguration);
}
pulsarSslFactory.update();
} catch (Exception e) {
log.error("SSL Context is not initialized", e);
PulsarSslConfiguration sslConfiguration = buildSslConfiguration(conf);
this.pulsarSslFactory.initialize(sslConfiguration);
log.error("Failed to refresh SSL context", e);
}
this.pulsarSslFactory.update();
} catch (Exception e) {
log.error("Failed to refresh SSL context", e);
}
});
}

}
Expand Down
2 changes: 1 addition & 1 deletion pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientInitializationTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,6 @@ public void testInitializeAuthWithTls() throws PulsarClientException {
.build();

verify(auth).start();
verify(auth, times(1)).getAuthData();
verify(auth, times(0)).getAuthData();
}
}
32 changes: 19 additions & 13 deletions pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/AdminProxyHandler.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@
import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.web.AuthenticationFilter;
import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.client.api.AuthenticationDataProvider;
Expand Down Expand Up @@ -290,6 +291,19 @@ protected HttpClient newHttpClient() {
return new JettyHttpClient();
}

private String getWebServiceUrl() throws PulsarServerException {
if (isBlank(brokerWebServiceUrl)) {
ServiceLookupData availableBroker = discoveryProvider.nextBroker();
if (config.isTlsEnabledWithBroker()) {
return availableBroker.getWebServiceUrlTls();
} else {
return availableBroker.getWebServiceUrl();
}
} else {
return brokerWebServiceUrl;
}
}

@Override
protected String rewriteTarget(HttpServletRequest request) {
StringBuilder url = new StringBuilder();
Expand All @@ -305,17 +319,10 @@ protected String rewriteTarget(HttpServletRequest request) {

if (isFunctionsRestRequest && !isBlank(functionWorkerWebServiceUrl)) {
url.append(functionWorkerWebServiceUrl);
} else if (isBlank(brokerWebServiceUrl)) {
} else {
try {
ServiceLookupData availableBroker = discoveryProvider.nextBroker();

if (config.isTlsEnabledWithBroker()) {
url.append(availableBroker.getWebServiceUrlTls());
} else {
url.append(availableBroker.getWebServiceUrl());
}

if (LOG.isDebugEnabled()) {
url.append(getWebServiceUrl());
if (LOG.isDebugEnabled() && isBlank(brokerWebServiceUrl)) {
LOG.debug("[{}:{}] Selected active broker is {}", request.getRemoteAddr(), request.getRemotePort(),
url);
}
Expand All @@ -324,8 +331,6 @@ protected String rewriteTarget(HttpServletRequest request) {
request.getRemotePort(), e.getMessage(), e);
return null;
}
} else {
url.append(brokerWebServiceUrl);
}

if (url.lastIndexOf("/") == url.length() - 1) {
Expand Down Expand Up @@ -398,7 +403,8 @@ protected PulsarSslConfiguration buildSslConfiguration(AuthenticationDataProvide
protected PulsarSslFactory createPulsarSslFactory() {
try {
try {
AuthenticationDataProvider authData = proxyClientAuthentication.getAuthData();
AuthenticationDataProvider authData =
proxyClientAuthentication.getAuthData(URI.create(getWebServiceUrl()).getHost());
PulsarSslConfiguration pulsarSslConfiguration = buildSslConfiguration(authData);
PulsarSslFactory sslFactory =
(PulsarSslFactory) Class.forName(config.getBrokerClientSslFactoryPlugin())
Expand Down
Loading

0 comments on commit fd45029

Please sign in to comment.