Skip to content

Commit

Permalink
[PIP-158][improve][client] Split client TLS transport encryption from…
Browse files Browse the repository at this point in the history 
… authentication (apache#15634)
  • Loading branch information
@nodece
nodece authored Jul 26, 2022
1 parent 2330f6e commit 6f1f6aa
Show file tree
Hide file tree
Showing 12 changed files with 349 additions and 15 deletions.
64 changes: 63 additions & 1 deletion ...-broker/src/test/java/org/apache/pulsar/client/api/AuthenticatedProducerConsumerTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,15 +28,23 @@
import java.util.HashSet;
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import javax.crypto.SecretKey;
import javax.ws.rs.InternalServerErrorException;
import io.jsonwebtoken.SignatureAlgorithm;
import lombok.Cleanup;
import org.apache.pulsar.broker.authentication.AuthenticationProviderBasic;
import org.apache.pulsar.broker.authentication.AuthenticationProviderTls;
import org.apache.pulsar.broker.authentication.AuthenticationProviderToken;
import org.apache.pulsar.broker.authentication.utils.AuthTokenUtils;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.impl.auth.AuthenticationBasic;
import org.apache.pulsar.client.impl.auth.AuthenticationTls;
import org.apache.pulsar.client.impl.auth.AuthenticationToken;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.AuthAction;
Expand Down Expand Up @@ -64,6 +72,10 @@ public class AuthenticatedProducerConsumerTest extends ProducerConsumerBase {

private final String BASIC_CONF_FILE_PATH = "./src/test/resources/authentication/basic/.htpasswd";

private final SecretKey SECRET_KEY = AuthTokenUtils.createSecretKey(SignatureAlgorithm.HS256);
private final String ADMIN_TOKEN = AuthTokenUtils.createToken(SECRET_KEY, "admin", Optional.empty());


@BeforeMethod
@Override
protected void setup() throws Exception {
Expand Down Expand Up @@ -95,8 +107,15 @@ protected void setup() throws Exception {

Set<String> providers = new HashSet<>();
providers.add(AuthenticationProviderTls.class.getName());
providers.add(AuthenticationProviderBasic.class.getName());

System.setProperty("pulsar.auth.basic.conf", BASIC_CONF_FILE_PATH);
providers.add(AuthenticationProviderBasic.class.getName());

Properties properties = new Properties();
properties.setProperty("tokenSecretKey", AuthTokenUtils.encodeKeyBase64(SECRET_KEY));
conf.setProperties(properties);
providers.add(AuthenticationProviderToken.class.getName());

conf.setAuthenticationProviders(providers);

conf.setClusterName("test");
Expand Down Expand Up @@ -403,4 +422,47 @@ public void testDeleteAuthenticationPoliciesOfTopic() throws Exception {
admin.tenants().deleteTenant("p1");
admin.clusters().deleteCluster("test");
}

private final Authentication tlsAuth = new AuthenticationTls(TLS_CLIENT_CERT_FILE_PATH, TLS_CLIENT_KEY_FILE_PATH);
private final Authentication tokenAuth = new AuthenticationToken(ADMIN_TOKEN);

@DataProvider
public Object[][] tlsTransportWithAuth() {
Supplier<String> webServiceAddressTls = () -> pulsar.getWebServiceAddressTls();
Supplier<String> brokerServiceUrlTls = () -> pulsar.getBrokerServiceUrlTls();

return new Object[][]{
// Verify TLS transport encryption with TLS authentication
{webServiceAddressTls, tlsAuth},
{brokerServiceUrlTls, tlsAuth},
// Verify TLS transport encryption with token authentication
{webServiceAddressTls, tokenAuth},
{brokerServiceUrlTls, tokenAuth},
};
}

@Test(dataProvider = "tlsTransportWithAuth")
public void testTlsTransportWithAnyAuth(Supplier<String> url, Authentication auth) throws Exception {
final String topicName = "persistent://my-property/my-ns/my-topic-1";

internalSetup(new AuthenticationToken(ADMIN_TOKEN));
admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build());
admin.tenants().createTenant("my-property",
new TenantInfoImpl(Sets.newHashSet(), Sets.newHashSet("test")));
admin.namespaces().createNamespace("my-property/my-ns", Sets.newHashSet("test"));

@Cleanup
PulsarClient client = PulsarClient.builder().serviceUrl(url.get())
.tlsTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH)
.tlsTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH)
.tlsKeyFilePath(TLS_CLIENT_KEY_FILE_PATH)
.tlsCertificateFilePath(TLS_CLIENT_CERT_FILE_PATH)
.authentication(auth)
.allowTlsInsecureConnection(false)
.enableTlsHostnameVerification(false)
.build();

@Cleanup
Producer<byte[]> ignored = client.newProducer().topic(topicName).create();
}
}
45 changes: 42 additions & 3 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 @@ -27,17 +27,16 @@
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.impl.auth.AuthenticationTls;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.Assert;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;

import lombok.Cleanup;

@Test(groups = "broker-api")
public class TlsProducerConsumerTest extends TlsProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(TlsProducerConsumerTest.class);
Expand Down Expand Up @@ -252,4 +251,44 @@ private ByteArrayInputStream createByteInputStream(String filePath) throws IOExc
private ByteArrayInputStream getStream(AtomicInteger index, ByteArrayInputStream... streams) {
return streams[index.intValue()];
}

private final Authentication tlsAuth = new AuthenticationTls(TLS_CLIENT_CERT_FILE_PATH, TLS_CLIENT_KEY_FILE_PATH);

@DataProvider
public Object[] tlsTransport() {
Supplier<String> webServiceAddressTls = () -> pulsar.getWebServiceAddressTls();
Supplier<String> brokerServiceUrlTls = () -> pulsar.getBrokerServiceUrlTls();

return new Object[][]{
// Set TLS transport directly.
{webServiceAddressTls, null},
{brokerServiceUrlTls, null},
// Using TLS authentication data to set up TLS transport.
{webServiceAddressTls, tlsAuth},
{brokerServiceUrlTls, tlsAuth},
};
}

@Test(dataProvider = "tlsTransport")
public void testTlsTransport(Supplier<String> url, Authentication auth) throws Exception {
final String topicName = "persistent://my-property/my-ns/my-topic-1";

internalSetUpForNamespace();

ClientBuilder clientBuilder = PulsarClient.builder().serviceUrl(url.get())
.tlsTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH)
.allowTlsInsecureConnection(false)
.enableTlsHostnameVerification(false)
.authentication(auth);

if (auth == null) {
clientBuilder.tlsKeyFilePath(TLS_CLIENT_KEY_FILE_PATH).tlsCertificateFilePath(TLS_CLIENT_CERT_FILE_PATH);
}

@Cleanup
PulsarClient client = clientBuilder.build();

@Cleanup
Producer<byte[]> ignored = client.newProducer().topic(topicName).create();
}
}
64 changes: 63 additions & 1 deletion .../test/java/org/apache/pulsar/client/impl/KeyStoreTlsProducerConsumerTestWithAuthTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,18 +19,25 @@
package org.apache.pulsar.client.impl;

import static org.mockito.Mockito.spy;

import com.google.common.collect.Sets;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

import io.jsonwebtoken.SignatureAlgorithm;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.broker.authentication.AuthenticationProviderTls;
import org.apache.pulsar.broker.authentication.AuthenticationProviderToken;
import org.apache.pulsar.broker.authentication.utils.AuthTokenUtils;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.client.api.ClientBuilder;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
Expand All @@ -39,17 +46,23 @@
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.impl.auth.AuthenticationKeyStoreTls;
import org.apache.pulsar.client.impl.auth.AuthenticationToken;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;

import javax.crypto.SecretKey;

// TLS authentication and authorization based on KeyStore type config.
@Slf4j
@Test(groups = "broker-impl")
public class KeyStoreTlsProducerConsumerTestWithAuthTest extends ProducerConsumerBase {
private final SecretKey SECRET_KEY = AuthTokenUtils.createSecretKey(SignatureAlgorithm.HS256);
private final String CLIENTUSER_TOKEN = AuthTokenUtils.createToken(SECRET_KEY, "clientuser", Optional.empty());

private final String clusterName = "use";

Expand Down Expand Up @@ -92,6 +105,13 @@ protected void internalSetUpForBroker() {
conf.setAuthorizationEnabled(true);
Set<String> providers = new HashSet<>();
providers.add(AuthenticationProviderTls.class.getName());

Properties properties = new Properties();
properties.setProperty("tokenSecretKey", AuthTokenUtils.encodeKeyBase64(SECRET_KEY));
conf.setProperties(properties);

providers.add(AuthenticationProviderToken.class.getName());

conf.setAuthenticationProviders(providers);
conf.setNumExecutorThreadPoolSize(5);
}
Expand Down Expand Up @@ -255,4 +275,46 @@ public void testTlsClientAuthOverHTTPProtocol() throws Exception {
.subscribe();
}

private final Authentication tlsAuth =
new AuthenticationKeyStoreTls(KEYSTORE_TYPE, CLIENT_KEYSTORE_FILE_PATH, CLIENT_KEYSTORE_PW);
private final Authentication tokenAuth = new AuthenticationToken(CLIENTUSER_TOKEN);

@DataProvider
public Object[][] keyStoreTlsTransportWithAuth() {
Supplier<String> webServiceAddressTls = () -> pulsar.getWebServiceAddressTls();
Supplier<String> brokerServiceUrlTls = () -> pulsar.getBrokerServiceUrlTls();

return new Object[][]{
// Verify JKS TLS transport encryption with TLS authentication
{webServiceAddressTls, tlsAuth},
{brokerServiceUrlTls, tlsAuth},
// Verify JKS TLS transport encryption with token authentication
{webServiceAddressTls, tokenAuth},
{brokerServiceUrlTls, tokenAuth},
};
}

@Test(dataProvider = "keyStoreTlsTransportWithAuth")
public void testKeyStoreTlsTransportWithAuth(Supplier<String> url, Authentication auth) throws Exception {
final String topicName = "persistent://my-property/my-ns/my-topic-1";

internalSetUpForNamespace();

@Cleanup
PulsarClient client = PulsarClient.builder().serviceUrl(url.get())
.useKeyStoreTls(true)
.tlsTrustStoreType(KEYSTORE_TYPE)
.tlsTrustStorePath(BROKER_TRUSTSTORE_FILE_PATH)
.tlsTrustStorePassword(BROKER_TRUSTSTORE_PW)
.tlsKeyStoreType(KEYSTORE_TYPE)
.tlsKeyStorePath(CLIENT_KEYSTORE_FILE_PATH)
.tlsKeyStorePassword(CLIENT_KEYSTORE_PW)
.authentication(auth)
.allowTlsInsecureConnection(false)
.enableTlsHostnameVerification(false)
.build();

@Cleanup
Producer<byte[]> ignored = client.newProducer().topic(topicName).create();
}
}
50 changes: 49 additions & 1 deletion ...st/java/org/apache/pulsar/client/impl/KeyStoreTlsProducerConsumerTestWithoutAuthTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,16 +19,18 @@
package org.apache.pulsar.client.impl;

import static org.mockito.Mockito.spy;

import com.google.common.collect.Sets;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.client.api.ClientBuilder;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
Expand All @@ -42,6 +44,7 @@
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;

// TLS test without authentication and authorization based on KeyStore type config.
Expand Down Expand Up @@ -241,4 +244,49 @@ public void testTlsClientAuthOverHTTPProtocol() throws Exception {
}
}

private final Authentication tlsAuth =
new AuthenticationKeyStoreTls(KEYSTORE_TYPE, CLIENT_KEYSTORE_FILE_PATH, CLIENT_KEYSTORE_PW);

@DataProvider
public Object[][] keyStoreTlsTransport() {
Supplier<String> webServiceAddressTls = () -> pulsar.getWebServiceAddressTls();
Supplier<String> brokerServiceUrlTls = () -> pulsar.getBrokerServiceUrlTls();

return new Object[][]{
// Set TLS transport directly.
{webServiceAddressTls, null},
{brokerServiceUrlTls, null},
// Using TLS authentication data to set up TLS transport.
{webServiceAddressTls, tlsAuth},
{brokerServiceUrlTls, tlsAuth},
};
}

@Test(dataProvider = "keyStoreTlsTransport")
public void testKeyStoreTlsTransport(Supplier<String> url, Authentication auth) throws Exception {
final String topicName = "persistent://my-property/my-ns/my-topic-1";

internalSetUpForNamespace();

ClientBuilder clientBuilder = PulsarClient.builder().serviceUrl(url.get())
.useKeyStoreTls(true)
.tlsTrustStoreType(KEYSTORE_TYPE)
.tlsTrustStorePath(BROKER_TRUSTSTORE_FILE_PATH)
.tlsTrustStorePassword(BROKER_TRUSTSTORE_PW)
.allowTlsInsecureConnection(false)
.enableTlsHostnameVerification(false)
.authentication(auth);

if (auth == null) {
clientBuilder.tlsKeyStoreType(KEYSTORE_TYPE)
.tlsKeyStorePath(CLIENT_KEYSTORE_FILE_PATH)
.tlsKeyStorePassword(CLIENT_KEYSTORE_PW);
}

@Cleanup
PulsarClient client = clientBuilder.build();

@Cleanup
Producer<byte[]> ignored = client.newProducer().topic(topicName).create();
}
}
Loading

0 comments on commit 6f1f6aa

Please sign in to comment.