[improve][client][PIP-158] Split client TLS transport encryption from…

… authentication

Signed-off-by: Zixuan Liu <[email protected]>

pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthenticatedProducerConsumerTest.java

@@ -28,15 +28,20 @@
 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.ws.rs.InternalServerErrorException;
+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.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;
@@ -64,6 +69,10 @@ public class AuthenticatedProducerConsumerTest extends ProducerConsumerBase {
 
     private final String BASIC_CONF_FILE_PATH = "./src/test/resources/authentication/basic/.htpasswd";
 
+    private final String JWT_SECRET_PATH = "./src/test/resources/authentication/jwt/my-secret.key";
+
+    private final String ADMIN_TOKEN = "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJhZG1pbiJ9.WgydImo6Rm2w9Jo4ABIEcqlr6HTBtThCVJlGBMyV8Hw";
+
     @BeforeMethod
     @Override
     protected void setup() throws Exception {
@@ -95,8 +104,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", JWT_SECRET_PATH);
+        conf.setProperties(properties);
+        providers.add(AuthenticationProviderToken.class.getName());
+
         conf.setAuthenticationProviders(providers);
 
         conf.setClusterName("test");
@@ -403,4 +419,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();
+    }
 }

pulsar-broker/src/test/java/org/apache/pulsar/client/api/TlsProducerConsumerTest.java

@@ -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);
@@ -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();
+    }
 }

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/KeyStoreTlsProducerConsumerTestWithAuthTest.java

@@ -19,18 +19,22 @@
 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 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.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;
@@ -39,17 +43,22 @@
 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;
 
 // TLS authentication and authorization based on KeyStore type config.
 @Slf4j
 @Test(groups = "broker-impl")
 public class KeyStoreTlsProducerConsumerTestWithAuthTest extends ProducerConsumerBase {
+    private final String JWT_SECRET_PATH = "./src/test/resources/authentication/jwt/my-secret.key";
+
+    private final String CLIENTUSER_TOKEN = "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJjbGllbnR1c2VyIn0.GDeH_gwktjoX4hdGqT_47nKkrfZdZ1ol5S2k0prAW_o";
 
     private final String clusterName = "use";
 
@@ -92,6 +101,12 @@ protected void internalSetUpForBroker() {
         conf.setAuthorizationEnabled(true);
         Set<String> providers = new HashSet<>();
         providers.add(AuthenticationProviderTls.class.getName());
+
+        Properties properties = new Properties();
+        properties.setProperty("tokenSecretKey", JWT_SECRET_PATH);
+        conf.setProperties(properties);
+        providers.add(AuthenticationProviderToken.class.getName());
+
         conf.setAuthenticationProviders(providers);
         conf.setNumExecutorThreadPoolSize(5);
     }
@@ -255,4 +270,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();
+    }
 }

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/KeyStoreTlsProducerConsumerTestWithoutAuthTest.java

@@ -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;
@@ -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.
@@ -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();
+    }
 }

pulsar-broker/src/test/resources/authentication/jwt/README.md

@@ -0,0 +1,32 @@
+<!--
+
+    Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+-->
+
+### Create a secret key
+
+```shell
+bin/pulsar tokens create-secret-key --output my-secret.key
+```
+
+### Create a token
+
+```shell
+bin/pulsar tokens create --secret-keymy-secret.key --subject admin
+```

pulsar-broker/src/test/resources/authentication/jwt/my-secret.key

@@ -0,0 +1,2 @@
+���"}�>�K0'֪=�.�)���?
+��GOu�