From 7b2a14c0d7faaca16f46d2c10cca89080ef66d1d Mon Sep 17 00:00:00 2001
From: Michael Marshall <mmarshall@apache.org>
Date: Wed, 15 Feb 2023 23:05:03 -0600
Subject: [PATCH] [improve][broker] Require authRole is proxyRole to set
 originalPrincipal (#19455)

Co-authored-by: Lari Hotari <lhotari@apache.org>
(cherry picked from commit aa63a5567a9e5d466b311a54d5dcc2cb05c2b5cd)
---
 .../authorization/AuthorizationService.java   |  67 ++++++++++++------
 .../admin/impl/PersistentTopicsBase.java      |   2 +-
 .../pulsar/broker/service/ServerCnx.java      |  31 +-------
 .../pulsar/broker/web/PulsarWebResource.java  |  27 +++----
 .../pulsar/broker/auth/AuthorizationTest.java |  41 ++++++++++-
 .../pulsar/broker/service/ServerCnxTest.java  |   4 ++
 .../impl/AdminApiKeyStoreTlsAuthTest.java     |  22 +++---
 .../pulsar/client/impl/KeyStoreTlsTest.java   |   8 +--
 .../keystoretls/broker.keystore.jks           | Bin 3723 -> 2254 bytes
 .../keystoretls/broker.truststore.jks         | Bin 838 -> 969 bytes
 .../broker.truststore.nopassword.jks          | Bin 0 -> 969 bytes
 .../keystoretls/client.keystore.jks           | Bin 3726 -> 2257 bytes
 .../keystoretls/client.truststore.jks         | Bin 838 -> 971 bytes
 .../client.truststore.nopassword.jks          | Bin 0 -> 971 bytes
 .../proxy-and-client.truststore.jks           | Bin 0 -> 1891 bytes
 .../keystoretls/proxy.keystore.jks            | Bin 0 -> 2245 bytes
 .../keystoretls/proxy.truststore.jks          | Bin 0 -> 971 bytes
 ...roxyAuthenticatedProducerConsumerTest.java |  44 ++++++++----
 .../server/ProxyWithAuthorizationNegTest.java |   2 +
 .../server/ProxyWithJwtAuthorizationTest.java |  13 ++--
 20 files changed, 160 insertions(+), 101 deletions(-)
 create mode 100644 pulsar-broker/src/test/resources/authentication/keystoretls/broker.truststore.nopassword.jks
 create mode 100644 pulsar-broker/src/test/resources/authentication/keystoretls/client.truststore.nopassword.jks
 create mode 100644 pulsar-broker/src/test/resources/authentication/keystoretls/proxy-and-client.truststore.jks
 create mode 100644 pulsar-broker/src/test/resources/authentication/keystoretls/proxy.keystore.jks
 create mode 100644 pulsar-broker/src/test/resources/authentication/keystoretls/proxy.truststore.jks

diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java
index 3baaf57990a9a6..05f146e8953888 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/AuthorizationService.java
@@ -19,10 +19,10 @@
 package org.apache.pulsar.broker.authorization;
 
 import static java.util.concurrent.TimeUnit.SECONDS;
+import java.net.SocketAddress;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
-import javax.ws.rs.core.Response;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.broker.PulsarServerException;
 import org.apache.pulsar.broker.ServiceConfiguration;
@@ -37,7 +37,6 @@
 import org.apache.pulsar.common.policies.data.TenantInfo;
 import org.apache.pulsar.common.policies.data.TenantOperation;
 import org.apache.pulsar.common.policies.data.TopicOperation;
-import org.apache.pulsar.common.util.FutureUtil;
 import org.apache.pulsar.common.util.RestException;
 import org.apache.pulsar.metadata.api.MetadataStoreException;
 import org.slf4j.Logger;
@@ -293,19 +292,39 @@ public CompletableFuture<Boolean> allowSinkOpsAsync(NamespaceName namespaceName,
         return provider.allowSinkOpsAsync(namespaceName, role, authenticationData);
     }
 
-    private static void validateOriginalPrincipal(Set<String> proxyRoles, String authenticatedPrincipal,
-                                                  String originalPrincipal) {
-        if (proxyRoles.contains(authenticatedPrincipal)) {
-            // Request has come from a proxy
+    public boolean isValidOriginalPrincipal(String authenticatedPrincipal,
+                                            String originalPrincipal,
+                                            AuthenticationDataSource authDataSource) {
+        SocketAddress remoteAddress = authDataSource != null ? authDataSource.getPeerAddress() : null;
+        return isValidOriginalPrincipal(authenticatedPrincipal, originalPrincipal, remoteAddress);
+    }
+
+    /**
+     * Validates that the authenticatedPrincipal and the originalPrincipal are a valid combination.
+     * Valid combinations fulfill the following rule: the authenticatedPrincipal is in
+     * {@link ServiceConfiguration#getProxyRoles()}, if, and only if, the originalPrincipal is set to a role
+     * that is not also in {@link ServiceConfiguration#getProxyRoles()}.
+     * @return true when roles are a valid combination and false when roles are an invalid combination
+     */
+    public boolean isValidOriginalPrincipal(String authenticatedPrincipal,
+                                            String originalPrincipal,
+                                            SocketAddress remoteAddress) {
+        String errorMsg = null;
+        if (conf.getProxyRoles().contains(authenticatedPrincipal)) {
             if (StringUtils.isBlank(originalPrincipal)) {
-                log.warn("Original principal empty in request authenticated as {}", authenticatedPrincipal);
-                throw new RestException(Response.Status.UNAUTHORIZED, "Original principal cannot be empty if the "
-                        + "request is via proxy.");
-            }
-            if (proxyRoles.contains(originalPrincipal)) {
-                log.warn("Original principal {} cannot be a proxy role ({})", originalPrincipal, proxyRoles);
-                throw new RestException(Response.Status.UNAUTHORIZED, "Original principal cannot be a proxy role");
+                errorMsg = "originalPrincipal must be provided when connecting with a proxy role.";
+            } else if (conf.getProxyRoles().contains(originalPrincipal)) {
+                errorMsg = "originalPrincipal cannot be a proxy role.";
             }
+        } else if (StringUtils.isNotBlank(originalPrincipal)) {
+            errorMsg = "cannot specify originalPrincipal when connecting without valid proxy role.";
+        }
+        if (errorMsg != null) {
+            log.warn("[{}] Illegal combination of role [{}] and originalPrincipal [{}]: {}", remoteAddress,
+                    authenticatedPrincipal, originalPrincipal, errorMsg);
+            return false;
+        } else {
+            return true;
         }
     }
 
@@ -340,7 +359,9 @@ public CompletableFuture<Boolean> allowTenantOperationAsync(String tenantName,
                                                                 String originalRole,
                                                                 String role,
                                                                 AuthenticationDataSource authData) {
-        validateOriginalPrincipal(conf.getProxyRoles(), role, originalRole);
+        if (!isValidOriginalPrincipal(role, originalRole, authData)) {
+            return CompletableFuture.completedFuture(false);
+        }
         if (isProxyRole(role)) {
             CompletableFuture<Boolean> isRoleAuthorizedFuture = allowTenantOperationAsync(
                     tenantName, operation, role, authData);
@@ -396,7 +417,9 @@ public CompletableFuture<Boolean> allowNamespaceOperationAsync(NamespaceName nam
                                                                    String originalRole,
                                                                    String role,
                                                                    AuthenticationDataSource authData) {
-        validateOriginalPrincipal(conf.getProxyRoles(), role, originalRole);
+        if (!isValidOriginalPrincipal(role, originalRole, authData)) {
+            return CompletableFuture.completedFuture(false);
+        }
         if (isProxyRole(role)) {
             CompletableFuture<Boolean> isRoleAuthorizedFuture = allowNamespaceOperationAsync(
                     namespaceName, operation, role, authData);
@@ -438,7 +461,9 @@ public CompletableFuture<Boolean> allowNamespacePolicyOperationAsync(NamespaceNa
                                                                          String originalRole,
                                                                          String role,
                                                                          AuthenticationDataSource authData) {
-        validateOriginalPrincipal(conf.getProxyRoles(), role, originalRole);
+        if (!isValidOriginalPrincipal(role, originalRole, authData)) {
+            return CompletableFuture.completedFuture(false);
+        }
         if (isProxyRole(role)) {
             CompletableFuture<Boolean> isRoleAuthorizedFuture = allowNamespacePolicyOperationAsync(
                     namespaceName, policy, operation, role, authData);
@@ -495,10 +520,8 @@ public CompletableFuture<Boolean> allowTopicPolicyOperationAsync(TopicName topic
                                                                      String originalRole,
                                                                      String role,
                                                                      AuthenticationDataSource authData) {
-        try {
-            validateOriginalPrincipal(conf.getProxyRoles(), role, originalRole);
-        } catch (RestException e) {
-            return FutureUtil.failedFuture(e);
+        if (!isValidOriginalPrincipal(role, originalRole, authData)) {
+            return CompletableFuture.completedFuture(false);
         }
         if (isProxyRole(role)) {
             CompletableFuture<Boolean> isRoleAuthorizedFuture = allowTopicPolicyOperationAsync(
@@ -582,7 +605,9 @@ public CompletableFuture<Boolean> allowTopicOperationAsync(TopicName topicName,
                                                                String originalRole,
                                                                String role,
                                                                AuthenticationDataSource authData) {
-        validateOriginalPrincipal(conf.getProxyRoles(), role, originalRole);
+        if (!isValidOriginalPrincipal(role, originalRole, authData)) {
+            return CompletableFuture.completedFuture(false);
+        }
         if (isProxyRole(role)) {
             CompletableFuture<Boolean> isRoleAuthorizedFuture = allowTopicOperationAsync(
                     topicName, operation, role, authData);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
index 1ed27d4a757e10..bd55067b563a90 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -3991,7 +3991,7 @@ protected void internalOffloadStatus(AsyncResponse asyncResponse, boolean author
                 });
     }
 
-    public static CompletableFuture<PartitionedTopicMetadata> getPartitionedTopicMetadata(
+    public CompletableFuture<PartitionedTopicMetadata> getPartitionedTopicMetadata(
             PulsarService pulsar, String clientAppId, String originalPrincipal,
             AuthenticationDataSource authenticationData, TopicName topicName) {
         CompletableFuture<PartitionedTopicMetadata> metadataFuture = new CompletableFuture<>();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
index 899bdd496268da..1e7ccfb298a653 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
@@ -188,7 +188,6 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
     private int nonPersistentPendingMessages = 0;
     private final int maxNonPersistentPendingMessages;
     private String originalPrincipal = null;
-    private Set<String> proxyRoles;
     private boolean authenticateOriginalAuthData;
     private final boolean schemaValidationEnforced;
     private String authMethod = "none";
@@ -261,7 +260,6 @@ public ServerCnx(PulsarService pulsar, String listenerName) {
         this.recentlyClosedProducers = new HashMap<>();
         this.replicatorPrefix = conf.getReplicatorPrefix();
         this.maxNonPersistentPendingMessages = conf.getMaxConcurrentNonPersistentMessagePerConnection();
-        this.proxyRoles = conf.getProxyRoles();
         this.authenticateOriginalAuthData = conf.isAuthenticateOriginalAuthData();
         this.schemaValidationEnforced = conf.isSchemaValidationEnforced();
         this.maxMessageSize = conf.getMaxMessageSize();
@@ -367,32 +365,6 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E
         ctx.close();
     }
 
-    /**
-     * When transitioning from Connecting to Connected, this method validates the roles.
-     * If the authRole is one of proxyRoles, the following must be true:
-     * - the originalPrincipal is given while connecting
-     * - originalPrincipal is not blank
-     * - originalPrincipal is not a proxy principal.
-     * @return true when roles are valid and false when roles are invalid
-     */
-    private boolean isValidRoleAndOriginalPrincipal() {
-        String errorMsg = null;
-        if (proxyRoles.contains(authRole)) {
-            if (StringUtils.isBlank(originalPrincipal)) {
-                errorMsg = "originalPrincipal must be provided when connecting with a proxy role.";
-            } else if (proxyRoles.contains(originalPrincipal)) {
-                errorMsg = "originalPrincipal cannot be a proxy role.";
-            }
-        }
-        if (errorMsg != null) {
-            log.warn("[{}] Illegal combination of role [{}] and originalPrincipal [{}]: {}", remoteAddress, authRole,
-                    originalPrincipal, errorMsg);
-            return false;
-        } else {
-            return true;
-        }
-    }
-
     // ////
     // // Incoming commands handling
     // ////
@@ -685,7 +657,8 @@ private void doAuthentication(AuthData clientData,
             if (state != State.Connected) {
                 // First time authentication is done
                 if (service.isAuthenticationEnabled() && service.isAuthorizationEnabled()) {
-                    if (!isValidRoleAndOriginalPrincipal()) {
+                    if (!service.getAuthorizationService()
+                            .isValidOriginalPrincipal(this.authRole, originalPrincipal, remoteAddress)) {
                         state = State.Failed;
                         service.getPulsarStats().recordConnectionCreateFail();
                         final ByteBuf msg = Commands.newError(-1, ServerError.AuthorizationError, "Invalid roles.");
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
index 39211aca7ad9a1..67904b5ff563ad 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/web/PulsarWebResource.java
@@ -149,19 +149,11 @@ public static boolean isClientAuthenticated(String appId) {
         return appId != null;
     }
 
-    private static void validateOriginalPrincipal(Set<String> proxyRoles, String authenticatedPrincipal,
-                                                  String originalPrincipal) {
-        if (proxyRoles.contains(authenticatedPrincipal)) {
-            // Request has come from a proxy
-            if (StringUtils.isBlank(originalPrincipal)) {
-                log.warn("Original principal empty in request authenticated as {}", authenticatedPrincipal);
-                throw new RestException(Status.UNAUTHORIZED,
-                        "Original principal cannot be empty if the request is via proxy.");
-            }
-            if (proxyRoles.contains(originalPrincipal)) {
-                log.warn("Original principal {} cannot be a proxy role ({})", originalPrincipal, proxyRoles);
-                throw new RestException(Status.UNAUTHORIZED, "Original principal cannot be a proxy role");
-            }
+    private void validateOriginalPrincipal(String authenticatedPrincipal, String originalPrincipal) {
+        if (!pulsar.getBrokerService().getAuthorizationService()
+                .isValidOriginalPrincipal(authenticatedPrincipal, originalPrincipal, clientAuthData())) {
+            throw new RestException(Status.UNAUTHORIZED,
+                    "Invalid combination of Original principal cannot be empty if the request is via proxy.");
         }
     }
 
@@ -184,7 +176,7 @@ public CompletableFuture<Void> validateSuperUserAccessAsync(){
                     isClientAuthenticated(appId), appId);
         }
         String originalPrincipal = originalPrincipal();
-        validateOriginalPrincipal(pulsar.getConfiguration().getProxyRoles(), appId, originalPrincipal);
+        validateOriginalPrincipal(appId, originalPrincipal);
 
         if (pulsar.getConfiguration().getProxyRoles().contains(appId)) {
             BrokerService brokerService = pulsar.getBrokerService();
@@ -259,7 +251,7 @@ protected void validateAdminAccessForTenant(String tenant) {
         }
     }
 
-    protected static void validateAdminAccessForTenant(PulsarService pulsar, String clientAppId,
+    protected void validateAdminAccessForTenant(PulsarService pulsar, String clientAppId,
                                                 String originalPrincipal, String tenant,
                                                 AuthenticationDataSource authenticationData,
                                                 long timeout, TimeUnit unit) {
@@ -286,7 +278,7 @@ protected CompletableFuture<Void> validateAdminAccessForTenantAsync(String tenan
                 clientAuthData());
     }
 
-    protected static CompletableFuture<Void> validateAdminAccessForTenantAsync(
+    protected CompletableFuture<Void> validateAdminAccessForTenantAsync(
             PulsarService pulsar, String clientAppId,
             String originalPrincipal, String tenant,
             AuthenticationDataSource authenticationData) {
@@ -305,8 +297,7 @@ protected static CompletableFuture<Void> validateAdminAccessForTenantAsync(
                         if (!isClientAuthenticated(clientAppId)) {
                             throw new RestException(Status.FORBIDDEN, "Need to authenticate to perform the request");
                         }
-                        validateOriginalPrincipal(pulsar.getConfiguration().getProxyRoles(), clientAppId,
-                                originalPrincipal);
+                        validateOriginalPrincipal(clientAppId, originalPrincipal);
                         if (pulsar.getConfiguration().getProxyRoles().contains(clientAppId)) {
                             AuthorizationService authorizationService =
                                     pulsar.getBrokerService().getAuthorizationService();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/AuthorizationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/AuthorizationTest.java
index 39a91f72dc7423..ad69180b236dc0 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/AuthorizationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/AuthorizationTest.java
@@ -22,8 +22,14 @@
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertTrue;
 import static org.testng.Assert.fail;
+import com.google.common.collect.Sets;
+import java.net.SocketAddress;
+import java.util.Collections;
 import java.util.EnumSet;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
 import org.apache.pulsar.broker.authorization.AuthorizationService;
+import org.apache.pulsar.broker.resources.PulsarResources;
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.admin.PulsarAdminBuilder;
 import org.apache.pulsar.common.naming.TopicDomain;
@@ -32,11 +38,11 @@
 import org.apache.pulsar.common.policies.data.ClusterData;
 import org.apache.pulsar.common.policies.data.TenantInfoImpl;
 import org.apache.pulsar.common.policies.data.SubscriptionAuthMode;
+import org.mockito.Mockito;
 import org.testng.Assert;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
-import com.google.common.collect.Sets;
 
 @Test(groups = "flaky")
 public class AuthorizationTest extends MockedPulsarServiceBaseTest {
@@ -229,6 +235,39 @@ public void simple() throws Exception {
         admin.clusters().deleteCluster("c1");
     }
 
+    @Test
+    public void testOriginalRoleValidation() throws Exception {
+        ServiceConfiguration conf = new ServiceConfiguration();
+        conf.setProxyRoles(Collections.singleton("proxy"));
+        AuthorizationService auth = new AuthorizationService(conf, Mockito.mock(PulsarResources.class));
+
+        // Original principal should be supplied when authenticatedPrincipal is proxy role
+        assertTrue(auth.isValidOriginalPrincipal("proxy", "client", (SocketAddress) null));
+
+        // Non proxy role should not supply originalPrincipal
+        assertTrue(auth.isValidOriginalPrincipal("client", "", (SocketAddress) null));
+        assertTrue(auth.isValidOriginalPrincipal("client", null, (SocketAddress) null));
+
+        // Only likely in cases when authentication is disabled, but we still define these to be valid.
+        assertTrue(auth.isValidOriginalPrincipal(null, null, (SocketAddress) null));
+        assertTrue(auth.isValidOriginalPrincipal(null, "", (SocketAddress) null));
+        assertTrue(auth.isValidOriginalPrincipal("", null, (SocketAddress) null));
+        assertTrue(auth.isValidOriginalPrincipal("", "", (SocketAddress) null));
+
+        // Proxy role must supply an original principal
+        assertFalse(auth.isValidOriginalPrincipal("proxy", "", (SocketAddress) null));
+        assertFalse(auth.isValidOriginalPrincipal("proxy", null, (SocketAddress) null));
+
+        // OriginalPrincipal cannot be proxy role
+        assertFalse(auth.isValidOriginalPrincipal("proxy", "proxy", (SocketAddress) null));
+        assertFalse(auth.isValidOriginalPrincipal("client", "proxy", (SocketAddress) null));
+        assertFalse(auth.isValidOriginalPrincipal("", "proxy", (SocketAddress) null));
+        assertFalse(auth.isValidOriginalPrincipal(null, "proxy", (SocketAddress) null));
+
+        // Must gracefully handle a missing AuthenticationDataSource
+        assertTrue(auth.isValidOriginalPrincipal("proxy", "client", (AuthenticationDataSource) null));
+    }
+
     @Test
     public void testGetListWithGetBundleOp() throws Exception {
         String tenant = "p1";
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
index 34cc3ba2232d4c..f55cc387705d44 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java
@@ -416,6 +416,10 @@ public void testConnectCommandWithInvalidRoleCombinations() throws Exception {
         verifyAuthRoleAndOriginalPrincipalBehavior(authMethodName, "pass.proxy", "pass.proxy");
         verifyAuthRoleAndOriginalPrincipalBehavior(authMethodName, "pass.proxy", "");
         verifyAuthRoleAndOriginalPrincipalBehavior(authMethodName, "pass.proxy", null);
+        // Invalid combinations where original principal is set to a pass.proxy role
+        verifyAuthRoleAndOriginalPrincipalBehavior(authMethodName, "pass.client", "pass.proxy");
+        // Invalid combinations where the original principal is set to a non-proxy role
+        verifyAuthRoleAndOriginalPrincipalBehavior(authMethodName, "pass.client1", "pass.client");
     }
 
     private void verifyAuthRoleAndOriginalPrincipalBehavior(String authMethodName, String authData,
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/AdminApiKeyStoreTlsAuthTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/AdminApiKeyStoreTlsAuthTest.java
index 77533949b8722f..fe734e06008bd8 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/AdminApiKeyStoreTlsAuthTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/AdminApiKeyStoreTlsAuthTest.java
@@ -47,7 +47,6 @@
 import org.apache.pulsar.client.api.ProducerConsumerBase;
 import org.apache.pulsar.client.impl.auth.AuthenticationKeyStoreTls;
 import org.apache.pulsar.common.policies.data.ClusterData;
-import org.apache.pulsar.common.tls.NoopHostnameVerifier;
 import org.apache.pulsar.common.policies.data.TenantInfoImpl;
 import org.apache.pulsar.common.util.keystoretls.KeyStoreSSLContext;
 import org.glassfish.jersey.client.ClientConfig;
@@ -74,8 +73,14 @@ public class AdminApiKeyStoreTlsAuthTest extends ProducerConsumerBase {
             "./src/test/resources/authentication/keystoretls/client.keystore.jks";
     protected final String CLIENT_TRUSTSTORE_FILE_PATH =
             "./src/test/resources/authentication/keystoretls/client.truststore.jks";
+    protected final String PROXY_KEYSTORE_FILE_PATH =
+            "./src/test/resources/authentication/keystoretls/proxy.keystore.jks";
+    protected final String PROXY_AND_CLIENT_TRUSTSTORE_FILE_PATH =
+            "./src/test/resources/authentication/keystoretls/proxy-and-client.truststore.jks";
     protected final String CLIENT_KEYSTORE_PW = "111111";
     protected final String CLIENT_TRUSTSTORE_PW = "111111";
+    protected final String PROXY_KEYSTORE_PW = "111111";
+    protected final String PROXY_AND_CLIENT_TRUSTSTORE_PW = "111111";
 
     protected final String CLIENT_KEYSTORE_CN = "clientuser";
     protected final String KEYSTORE_TYPE = "JKS";
@@ -96,8 +101,8 @@ public void setup() throws Exception {
         conf.setTlsKeyStorePassword(BROKER_KEYSTORE_PW);
 
         conf.setTlsTrustStoreType(KEYSTORE_TYPE);
-        conf.setTlsTrustStore(CLIENT_TRUSTSTORE_FILE_PATH);
-        conf.setTlsTrustStorePassword(CLIENT_TRUSTSTORE_PW);
+        conf.setTlsTrustStore(PROXY_AND_CLIENT_TRUSTSTORE_FILE_PATH);
+        conf.setTlsTrustStorePassword(PROXY_AND_CLIENT_TRUSTSTORE_PW);
 
         conf.setClusterName(clusterName);
         conf.setTlsRequireTrustedClientCertOnConnect(true);
@@ -107,6 +112,7 @@ public void setup() throws Exception {
 
         // config for authentication and authorization.
         conf.setSuperUserRoles(Sets.newHashSet(CLIENT_KEYSTORE_CN));
+        conf.setProxyRoles(Sets.newHashSet("proxy"));
         conf.setAuthenticationEnabled(true);
         conf.setAuthorizationEnabled(true);
         Set<String> providers = new HashSet<>();
@@ -147,13 +153,13 @@ WebTarget buildWebClient() throws Exception {
 
         SSLContext sslCtx = KeyStoreSSLContext.createClientSslContext(
                 KEYSTORE_TYPE,
-                CLIENT_KEYSTORE_FILE_PATH,
-                CLIENT_KEYSTORE_PW,
+                PROXY_KEYSTORE_FILE_PATH,
+                PROXY_KEYSTORE_PW,
                 KEYSTORE_TYPE,
                 BROKER_TRUSTSTORE_FILE_PATH,
                 BROKER_TRUSTSTORE_PW);
 
-        clientBuilder.sslContext(sslCtx).hostnameVerifier(NoopHostnameVerifier.INSTANCE);
+        clientBuilder.sslContext(sslCtx);
         Client client = clientBuilder.build();
 
         return client.target(brokerUrlTls.toString());
@@ -186,11 +192,11 @@ public void testSuperUserCanListTenants() throws Exception {
     }
 
     @Test
-    public void testSuperUserCantListNamespaces() throws Exception {
+    public void testSuperUserCanListNamespaces() throws Exception {
         try (PulsarAdmin admin = buildAdminClient()) {
             admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
             admin.tenants().createTenant("tenant1",
-                                         new TenantInfoImpl(ImmutableSet.of("proxy"),
+                                         new TenantInfoImpl(ImmutableSet.of(""),
                                                         ImmutableSet.of("test")));
             admin.namespaces().createNamespace("tenant1/ns1");
             Assert.assertTrue(admin.namespaces().getNamespaces("tenant1").contains("tenant1/ns1"));
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/KeyStoreTlsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/KeyStoreTlsTest.java
index 2e839b93f194f9..425e08cb91faba 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/KeyStoreTlsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/KeyStoreTlsTest.java
@@ -31,23 +31,23 @@ public class KeyStoreTlsTest {
     protected final String BROKER_KEYSTORE_FILE_PATH =
             "./src/test/resources/authentication/keystoretls/broker.keystore.jks";
     protected final String BROKER_TRUSTSTORE_FILE_PATH =
-            "./src/test/resources/authentication/keystoretls/broker.truststore.jks";
+            "./src/test/resources/authentication/keystoretls/client.truststore.jks";
     protected final String BROKER_KEYSTORE_PW = "111111";
     protected final String BROKER_TRUSTSTORE_PW = "111111";
 
     protected final String CLIENT_KEYSTORE_FILE_PATH =
             "./src/test/resources/authentication/keystoretls/client.keystore.jks";
     protected final String CLIENT_TRUSTSTORE_FILE_PATH =
-            "./src/test/resources/authentication/keystoretls/client.truststore.jks";
+            "./src/test/resources/authentication/keystoretls/broker.truststore.jks";
     protected final String CLIENT_KEYSTORE_PW = "111111";
     protected final String CLIENT_TRUSTSTORE_PW = "111111";
     protected final String KEYSTORE_TYPE = "JKS";
 
     protected final String BROKER_TRUSTSTORE_FILE_NPD_PATH =
-            "./src/test/resources/authentication/keystoretls/pulsar_server_trust_npd.jks";
+            "./src/test/resources/authentication/keystoretls/client.truststore.nopassword.jks";
 
     protected final String CLIENT_TRUSTSTORE_FILE_NPD_PATH =
-            "./src/test/resources/authentication/keystoretls/pulsar_client_trust_npd.jks";
+            "./src/test/resources/authentication/keystoretls/broker.truststore.nopassword.jks";
 
     public static final Provider BC_PROVIDER = getProvider();
 
diff --git a/pulsar-broker/src/test/resources/authentication/keystoretls/broker.keystore.jks b/pulsar-broker/src/test/resources/authentication/keystoretls/broker.keystore.jks
index 8ef2c6c60ba81623c6afd7a6b0611f0ee1147616..6f2df055f26add400c012950df1967fbc2e2581c 100644
GIT binary patch
delta 1995
zcmV;+2Q>JL9nKLb{_Xzl00002000010000100v@mZ);_80004oNmTu6000F6FoFdA
zlTiZ`f9X{m&X|u!c0CEb3Pv`FDm05j0yUQt1Su;4k|;4=>wgkS0;cbp>S2@gq;CQs
zb1pjSRpjuMwlpM^va;0+eJRC$zH5ZI*VHc2yG!6~#r{AmJ`DxhBoONC&+JmxnFey`
zd^N!yi^7IKe6RXDJjHNN`fA$A;sJ2XV6cQ)e|;uU{TK&(Gh1F)6e1dC1huKPE<ya^
z^|pI=$LFp+S4B~NElPkB%k4EWR#>6U+b5mV2Ir14jT6l{x)v&=%5=gsXniHfkt0)A
z1nsg$2K}Q&Y;Dg)uN+Ct(8C?A^P#Awbqi_5PD?y)x$j@|O!@g1EF-}T(C*Zu-;N)K
ze{ZcCh7!)Sy%EDUOJ7_8DjIastQF%WcTUa0k+6;&bovC1D^2Cogg_i%*K-!-lQH|b
zJbkoELK-uHDd<tbkn@|Z!&!J!lfY#(ne;9R$EaM!vG2TqpXW(G7S(i{GPHy!&!}@n
ze<}4UqgTRIzhH7Jo?MV*<)4Eg{sZuuf2dYVVW5@NDgD*Ah<%Kl4*)AGmMitI`ELSS
zF-{l8aT9|lA>C4%K}Z;78Ok-%yk+v!g}Lwg>qtKfv(XV*r1$$zDy0|U#SJf^;&{x8
zu^oloU>>Ye9VCbieuDvuN)F#GjO`_Iy=j3^q$+5#0xtFhX+V!-b@dX*gp{Ape~ar0
z0=HviJ{Wy4KoD5wIgy5%;1gHFWSASA9$D=~L0Uz)x%9@2d(ol~QLC?0&s<9}kSW-P
zo4te1`MIZrHnM`6=9BTPhM^x$diy_jyMM~`!;?<Jj)=*LWY+_adTm_%^guGgc{8TL
z1xl*il1iuZdzx?12l(k0ZcDlsf1r>VIj&>HT9z<voTuN%<W^ou2Mbb6HybQ0o#<0K
z2Qab#T+AAr9$!Z_)EgqdYUK|5i-z<9+ykv&=@UH7o2WNHZf<mZVnZ%=Xu|ObwGNpb
zkDd8edczB@{}KJafw>Y8WuSllXtH=NLMAzDVu^ZLqCku05DOx@V>4f@f8(iOt9<bW
zrpiDmX_;VqzPG+QnDcmz@(lu}LzljaRm!W4&<pr~I(xPyWZ`8D;32`mlErXZ7Np}t
z7O{eRT?3gCFiL<2Y05#ER*=S5GL(`fSU?(DAt0b;z7DGI1bo;9zDE}1Sr(f$TQ<VU
z$ENYKZtX+iqPq1mNBw{Ff8z1?SBSZpXBZVydb2IZ7R^skew<THIb7)+g7bt`6gl+w
zRk|fO?P6h>fj`fNh+#KdDc*H<;A>_}RL`qof143ptxUE_-E>g~@c>QUTVg5;^Sj#{
z4PW%4q)$a&q+eFKK;f#}lBCKkB|1;@7*fySr^qZ5LF|!qyyzzRe;JQKUc@=1XXP3I
z|FXzgGET8EqP)xmxRIVF7SY;a$B7|CsJQL?fIT#uV(PbpiAojikoe{BA=85Y?-5AW
zA4o({(Cu?NuL%tozA)iUDA*KJbD#mfPu?h~y_ZVuJZ{s-zX^Af*4D<HLkMO})f2~$
zc7n6Xj>}EWnNy1-e?s(gk8m>sT(`br>ZbC7AI0*~D84da<f7O_codQyZZ$cK2t5#G
ziw$bSFiLapptEK|M=)<Hnh5=idZUNGr1kqE)u4^%V4F$_Ltm0RpGAx!W*59o+*QBS
zf`wU`&xlA)d0}BzOW$SN++M~~v6%CD0000100mesH842<DgXn9FoFYuFoFVVpaTK{
z0s;vD@(;#1a|7y9FbxI?Duzgg_YDC73k3j^y9jY0AQufXGcYnSGcYwZGBYw-7!NWr
zGBYqSGB7YTH8L|Yle!3!f9SX0?lOp&4hI-P{n+9($bM9Yo7kg!M{en%jCAu4yQYIf
z5qHFzt5bAjM*e}1as!tOH+XLUZe(cdp0I1Kd1_GM9vwN~Q40i)iA&@HYvdBsK}Rf(
zbubhf?J|?5{&RiY;M9}Zj6_w^Q%Vdj`ITQ!%P-|Y#k@0!>)28be;HLa0x=JP3&Kw|
zW2Oe<JCi&FTw^jl!zNIcw~S|wp3$RzdsK!s2mTc@`)y9=OqZ21;@AX>;s8s9WRY;L
zB!~w`MSU5=!6gDv`1A^+dukEGZ@JnApe;RRM~*1c7>Y%^d3(Cm*eOj;+WoQ~>$X`N
z@yUwntbKCajpsvsD$~aT0|5X5qaiRKFdYU1RUHll76cTCI$52va;#+g^No4!RQ@2t
zgGnTlQ3@1)afs{3^74c=_p7`-&b|@($u_kfla|kB21@a^^~{&K&65v`W@3fQ{lQ~?
zIN1UWYXmrZ?1JVc^Ng8|(ICd_UUi<f@a(Ocb|OK&i0Q+BJ&v03O0(iH%HU`@jeO?Y
z-2Fs#W>wRtkL&v=hP|#nlb!Z<GTQ}BZZwh>A%)X_4li@liT7!440JV>jN|{t_gu79
zz)5`ou55@~MlrKuWMZ%loP5xPXxZaZ?KSvXZ=P3TOE!=9J~0*iyWdg7{gReh)m1)w
z`!6n96~637IN-^&M$24rO8&<gEQB1{6%zbF%zXUxI3o7C+wujM!6x;~!0oyg<}(a8
dItQ~BqsmG0lUFo*!2*A}t99X^`Ul+!MdpNstVI9-

literal 3723
zcmeH~`9IX#AIIl2o27}d4aQE?AkCPugzO?)V_%AUMYe{K8C$twY=bGuQXv#cWs8!f
z24l@mNl2lx6p<8Ce5W4w(On<kzu^9G9_Mjh=bZQBocH^j^M0PK#jQmU2n7CF;GQ1A
zWHJ>50>ZyJef9>yoK@7Y`#~TG2bKnbU}?a+0tgrYg9QaZ8<LLK#Zh%DFG6H^`F+%=
zUw2@4!nqU^%o5gi0>BO!2+N1#WlInTC<FmFGO%Ua#YzhARK;Nls(7qAR#R2gRd^@<
zdlUD2^WUb>0NFnq24Elv4d4Z_*TM~^0RX7JL^+uU`J`=$EQ_o*VhWZ>&Klhk6~0ho
zupP%YV_B7S4)0T+e?Ue0SI<_BM#3D+X*k~1`4V<H?;8#dSw7M;JEa*o&lx_OCN`lH
z2-+XaGXt|^evX%Qo88zDLS%6h8Y|AjG%o6uUuknPblDlStLeQ8Keap1DCt`z_K2y2
z;;z^Fd^!a?0w0`BBK37br4iW!*aPBJyldyx`R?r1D~{7oTBVKj-u15?5{Tq*MuXpd
zPU4<)ptxs*O7|mgFpjhm%frm{cDlUq8`K49VY1U?))LaH&07?#l3Hc`6J5_fJ+@Si
z6w)6WCvJu<2b8nE1no=Av$Xd0t~AyLLjVv^Xot1IO0az`!2?A>1r-h#>Om8=L&9&)
zQzO4n_*iy#PhzFMcgsOVu&C&NAHe&4fB*n&Lt$75yZW=++3{q%9RT4IQg}yaN;yp)
zi!}G@S-P&x?0j`*O+{CWPRE`Ube;~@7?rEe*XXR+@h-Bo?KM$s&a1xk9=ed+w|9SP
z#On#O%wB4K)Q+f%R;=cvt5I1R7Tj{&tglni#}OqapLsfE#wj=C?wNy<A0E1~+=uv;
zny-ou+G&1(F_W_G7tS>%jVzsBu&8$_z3qRrPSnM5R>w1(Im_VjL#!3)tYqu2g=!QG
z<z;gP*Hznk!G`%UogBWKo~sX1N2gq)45ov{iomu`0pr|{wx9Fwh6tbD7ZdQfq}Y5<
z1h?;n5QIYRp%GX|>Ox*XKuEciReu}DRw|OR7{wgSFBI-2a>Y!sZwdh7B9T2kNWNss
z?-%8Z90<fd?=<LFEDxMp2@XI3MKB-0KB`a}bk5kM0k7*wo`IoDOcDzep^?egQkq9b
zxiUM}M+GiiSJBLp-#j%?puaMP=x+|eV-y+`kB9pQ)SE;desRvUZFfmAXsUkY)j+R?
zq~}Ju9&X6X;!ywcb73_qZ%^FZ)UE<)e^URsi_f;BiRU#+sC)CmG#|~JT{Q5J0NV|L
zr}30AwKSz$L6D*gWewAq;)74pOaa6++`D|Np;*j#Lmh0{!F=#eb@wc@`-U`yHu9cv
z@YK{5#1voo6KP$iXxVQ1mdM4&Bj$^P#uWeuzT1-7qir$AaY0s$>!X4?{20>kOr&vm
zteOkSzyqi?zo{Tt-5`t4<4o-F1J2`P7gQscEF_L<g@*=fDj~s#_#qsh-c)TaB@@Ts
z#lW216@P`0_xAG=n|-TuRmis6`lm?blq%4Saq=CN6Ap!tpu%)?{TwBB@A{?UR^5i9
z9dBEXD1Ebr|0+k<ilcr00Nl&NsD=cIGkUHy^t%x-A(N8PsvLceu`6iPX1>a1%PFZ-
zV{n;jg^AnhyvHRr+P2JbQB8YGXMf3eg)295eegMr|8;RZ#oIEZK`(kUw_<mOT1394
z!&^7VLe^f2&1CQEmGhx+ipsfxygK(-RHfE~eW@*Bmv4#B@fUhr-adOWVozd~IZJuG
z^le$RsCb}d76(g=lxsbjzUN)UCC^s3;?MlSWj3at-#8I2!&pS+`?9Ymg;2W>=rjC3
z)@4v7^$B$s>z=(7vU9_PVJvkj)4Sqqp_GJV1P^I-K&qDi4O6y)P9@gtR@Q-C-`Lmf
zFH=*|#ggWM6pUriBj7#<4ckdN&N=rU7RS&|d|Wpk%<A|ARO%I#Y!W@R<-2dVPHBUx
zvs`T%w!AMcR^i0kLQN4tx1w8MB6pAo5xVvsje}sE6!rj`(6RO$Hss!+=jN2d3T+wV
zap*sPsQUP-;LS^qT9?XCK5PkjWGE;khBcgAAj$+81aWGohvgwb(Ked~7Sy`;nNm#N
zQ#_+qO36K+hD*hYk9&DjJ}s-J9$4or>8sBer;|}~4-((Jyd7YB^%TDVXU=6~VbhbJ
z?nz`)+w`mtJtdTmY>hQf0xz@dtHt~#Di?~c4cX)3Hj=eUh(#@;+Fc@JTmJjCCYmt<
z(xk1_5aE@vu3u1Sn^?|uXV}mcbJTRB!KJ5!aiP)zooR$y$y;(W+S---v3+6=?a?%+
z{n4e7<CWYDQHdM)L7T5qP@x!stv6On6li?ARqk(`6hF^%+Vy70u121(oWH$v?0GH~
z_&RD>u8EojGrtqrCuVAQ-_Lo(#23vMxr=etYk#fcI>FW93CUSaz?|r;A+Oz=K7O}o
zey}mgZSt^G7{Yt)R!*IHE%)htbhj6h$>L^738*TaGn1{(_2VI4Q^o|Zz^X!4cf{9F
z%xeGTOOaK&_uXTR$Xsx>BIK<~M2w6s*&enqDQi2Tq?v6+`0eEyZZ$$y`wp`rJRzFA
zH9tHo^C82wCZk2rN91F?y?(u&Qmylr3no-vt>AhL=g`>+g?Q&2xz8C$JBZuFs7I4;
zfUShJO5c;G+9V!z<%UXAP;mC_m-Kf6!W@}3%*&IVjFgf3-gOn>Y0Pp{!1w`{P(oM}
zt$+9X-sm}X(qxmObvtAmhxTJ8W7g1nl<R$;Rq1^_jK1l3vy3SA10sF`0~Z?@VC*El
z2o|ttOf|j{YO$PX!cvuAzp^Jo?cU!~%fIFQzoG)+uc$z9#RlS#?AMk}7jOX%hrkno
z<glL#=RXvVe<<AlP<H&G@MDqREkgwUc<I?&`Trck7&+2g$G<m#e)tJQQ6Z(~gf8CD
zD>NfB^#y*v8j~b%$=8j?;|9+t=T+?O$Ox})qUVazT`D#Om!yJq`yT!>I@limih&xz
z<h8L>dzA`?5y;r|T))QbG{zZjN%az%75(-}gOm0?ax#9;BWo)A$t}4bNuK<$)A@!u
zv6UC;g?*hSv+A8%S+B40_PJNj$)QELC?k9-Hug5Gi-v=$H;>W1J+d}ivnQVk(gAD6
z#t{Qzc5%|GKT3CBt=7XzW5Y(1g!I)h=?Ip4W}L~L^0RT-2AebWySldC>N4!~W9JGr
zA`gcb%iFz);kaVan5K=oBj6{3;>s0CayXUtgG2rn*B{7`6%;jp&gm-$HLObMZIY91
zf<&2GkQt<U7D8Aq8@D=2pB6f1IY4-7@G>oGro6oOOI@~Lk&m+E(BcaN`J{x5cI|lk
z+cLD%j!fG6`CY<Cj|9+c4dYd*!U|Vg2hZd7d^5@`8|V>(AeBr@;n5?DQKm!FnHR{l
zHCNvc^R9d3D{G})w|G^Ip3YjCD6c_ncFdPR9$a(F9G~|L`IZK=Q*|*BU4EF+eF2-B
znNfQ^i$}%}g+Y3)yOBBfo86Oj6k$++0tSEOeeoOp`em-q%>B8`obScw&Jl-R_1>IH
zzSAmM3p2NC*D6D8=WI`)ui+jR-{m!k>yk{&79*|P_z$B0gXo{1qW>V84bkm&thU^S
Tl-g3=&c!Sw=9zD=b^HGSLGvMD

diff --git a/pulsar-broker/src/test/resources/authentication/keystoretls/broker.truststore.jks b/pulsar-broker/src/test/resources/authentication/keystoretls/broker.truststore.jks
index 96f12a38a72270055983ec89ba2543b3795f449f..9c35356c540212f5f3841d77cbf90995b963ca04 100644
GIT binary patch
literal 969
zcmezO_TO6u1_mY|W(3n*B}JvhCB-HAMX5lcHqTI&lMJj8dZq@J3=GU|22IRO22D&^
z3z(T0nV2{kKJg#3C}w^YWWdYDsnzDu_MMlJk(-slAkR?1fRBwil!Z;0Jv1*nFTWfv
z!hs>ejv>N@A;OIi5i$?}nanKAnUkNKn3IuTTw<Ug&TC|BU}R`)U}|b)Y!oHIZ)6A*
zFf=lN3eep}O^iy&fx^hjz}&>h&tTBR$i>ve$jI<w`~7!D9W(gYB^>|Wcw};-E~IVt
zjm6dOc`q0Clzisjy{y?;u>8oZ)xjkxZvPtli<qZ#n^)xh&r8X8J#Rzyy2`A8hjOx(
z_XD|E`Z~RzFl9dxy6ot#(^qOBBK6j2^0I%$wRav|nS8UyCG=9T7mwc0sqy}&^`AN(
z-DBMG`bH3+WT+XFA%7$HA%D~4Wo(bFC)=>ZBpcZtRtuQ6y(hhI-o?dr)gf)B?EgfK
ze&_l<_nAJ`=+O<9?neyXttk@=)~R%`ySvs(9zLkb6!7B<*W&6d!Nd7GZ?P@VwM}*J
z)3_wj<+{7Fde_w(n!bLw{%(?cy)9Du<H@d9Yif(`^geg4y?mUBnUR5Uv7&*zfh;gK
zW%*ddSVTIkBIj-@T9fkob8qF_kbeq?n>|&K0}+_AfPu)!P}uSM_@_@Trr%fZu{pO-
z@aIXht#Xs6olj%)`nc`undv*vPUi1QOKLs+_h53J#Z4xjY!-{^H%-q}KljY+y{K^P
zbzJGZZ6DsOomHmfxVPiw;d<M?Ss%SNKQcJ=Aj7h^=GpB#e_cw`LN71xfBjpdZSOj}
z$#cJz8Qo^}%`=%Krr3I!Prvwb=l9G!o)XikJ&*q%`yR6;^nhnA!@8V~Xg9;nNhwJi
zc<0nyXvw(wIOwhEkEr~4;Yr?R{om~jMgQ!+A9&>Nq-l{?L+z@6>+3~{?t9~A@!;eZ
zx6?6&UjL3u>a@t*6czg8aHi(Z7Yn6tyKaACoqka5>*)h;cZoeS<}tTo-@N!#Na>;&
V#kJx!J=4krj<8hlYf4J=003<vaU%c#

literal 838
zcmezO_TO6u1_mY|W(3o0$%#ez`6WPZ#ozER-xydU^h^yb85o$^4Vsvl4VoBVFJNY3
zWMUE#X80)=wp2!J=cjw6iu^+9rX_#44S3l&wc0$|zVk9Na<eiR2paMOg_uKGn0eTo
z9fN@88pw(B8W|dx7#SOw8CVz@MTzqogSduJE;cPqj7rD`GqN%;H!<=v0NujH)Wpch
zuxo`*7oWhDqwn*T1U5VG6Imhm+<BFx_@r%)|62r~`E2U!Gfv+%-&Ri{A^C{;ev=o6
z!mqYQMa(exFz>G+8}o<gGtVDd<h|smcs^13u1y|;Wf9*qR-b*}T9sm-|NJS+GmXn+
z|N056=F{!h%{?0K6v<nla^RhwP|4{$=gz+y45HmbwN#!s2-+;*&fC-5nRD(0ivrKA
z3kJ3_CB{)F7QQ?^>&x7*M^}z0INQHI|J_r#hCM=x>Giixt_PvTaZ}0^&I>H>i9Qmr
zuH3_(H}YQQMLPy7%~=x_zqd`?>~&bpxATZnR(n+MwWRm!1VkM!-46I&{vmhWp<f01
z?el#6GSW7<STiv*GB7R<HV8D31%|aOABz}^h?@UmdzN<V(u$QYOKN@<3myvIQfMF#
zl2&GsFc53Nu7Dq;K$wy7KMSh?Gmt_Kc3^x0gPoDVxUHN2=Cd^%u9qbap0)Rh`u>df
zaou-4JFAWkgF=yrM@8m0l{e2fKe3+sb<LWiPXeT0r0iO=RcdkmIbF;C$|rX{rk*XC
zU(a2?{)mCagDB^<6AhRSr+S<_p_LvcCapTPy8Bu9oYF0IZgL;@#vY2hB&2<Cq2$G2
zi;t}PI%kEv>D%9V{eATt?_H5=R%gZRl#C2}Zj)TG?|BbjCeQaJHlJoWd@nO!aB1Ew
z&Z3>0gHl+p3Td8T&-k7EWl#Ujhf(#8kBX$1Fa?F@-sU>_|BjG4vv{?BWA5b@%e*uy
zxiTh|@~F-6yv};O|INJI+|qUOzUPm!|LYehep|oq;{3(pX9GAJAG$@erG7r;6}f8K
PmV=Leuw73+(aQ?}0!d1=

diff --git a/pulsar-broker/src/test/resources/authentication/keystoretls/broker.truststore.nopassword.jks b/pulsar-broker/src/test/resources/authentication/keystoretls/broker.truststore.nopassword.jks
new file mode 100644
index 0000000000000000000000000000000000000000..75c3fd8012f9625d6d5160f8905235c314b3ccdf
GIT binary patch
literal 969
zcmezO_TO6u1_mY|W(3n*B}JvhCB-HAMX5lcHqTI&lMJj8dZq@J3=GU|22IRO22D&^
z3z(T0nV2{kKJg#3C}w^YWWdYDsnzDu_MMlJk(-slAkR?1fRBwil!Z;0Jv1*nFTWfv
z!hs>ejv>N@A;OIi5i$?}nanKAnUkNKn3IuTTw<Ug&TC|BU}R`)U}|b)Y!oHIZ)6A*
zFf=lN3eep}O^iy&fx^hjz}&>h&tTBR$i>ve$jI<w`~7!D9W(gYB^>|Wcw};-E~IVt
zjm6dOc`q0Clzisjy{y?;u>8oZ)xjkxZvPtli<qZ#n^)xh&r8X8J#Rzyy2`A8hjOx(
z_XD|E`Z~RzFl9dxy6ot#(^qOBBK6j2^0I%$wRav|nS8UyCG=9T7mwc0sqy}&^`AN(
z-DBMG`bH3+WT+XFA%7$HA%D~4Wo(bFC)=>ZBpcZtRtuQ6y(hhI-o?dr)gf)B?EgfK
ze&_l<_nAJ`=+O<9?neyXttk@=)~R%`ySvs(9zLkb6!7B<*W&6d!Nd7GZ?P@VwM}*J
z)3_wj<+{7Fde_w(n!bLw{%(?cy)9Du<H@d9Yif(`^geg4y?mUBnUR5Uv7&*zfh;gK
zW%*ddSVTIkBIj-@T9fkob8qF_kbeq?n>|&K0}+_AfPu)!P}uSM_@_@Trr%fZu{pO-
z@aIXht#Xs6olj%)`nc`undv*vPUi1QOKLs+_h53J#Z4xjY!-{^H%-q}KljY+y{K^P
zbzJGZZ6DsOomHmfxVPiw;d<M?Ss%SNKQcJ=Aj7h^=GpB#e_cw`LN71xfBjpdZSOj}
z$#cJz8Qo^}%`=%Krr3I!Prvwb=l9G!o)XikJ&*q%`yR6;^nhnA!@8V~Xg9;nNhwJi
zc<0nyXvw(wIOwhEkEr~4;Yr?R{om~jMgQ!+A9&>Nq-l{?L+z@6>+3~{?t9~A@!;eZ
zx6?6&UjL3u>a@t*6czg8aHi(Z7Yn6tyKaACoqka5>*)h;cZoeS<}tTo-@N$L&kZ|{
V8ei?$#K_k1u!4R1v1Y|TWdM`hbm#y8

literal 0
HcmV?d00001

diff --git a/pulsar-broker/src/test/resources/authentication/keystoretls/client.keystore.jks b/pulsar-broker/src/test/resources/authentication/keystoretls/client.keystore.jks
index 375e2e00bb410f76c744f0f57bcfeee4a65a1eae..0c9d33408e1c1aa52971b2037af7893204a79876 100644
GIT binary patch
delta 2025
zcmV<F2Nw8_9nleg{_Xzl00002000010000100v`hX=QG70004oNmT!f000F7FoFdB
zFb)O^D+U1s0V)C!0RaU71cC(WXK<WHr2rI-jq(2*({8fA3^Sn%QDK~;<s*ykrZ;3X
z3Up}IAK(#k$dUP6my((a-$ld}Qypbl;9m{tlBd`g1f-*XWNX`<ht|q$G6;iR>0w53
z9}z1l$<#Ku0ATEdb!k>s7hQi{Q5mk-i!vI-9@JGxzK9Uj{wDYcd1RJ<BX)t2&~TeQ
z_s4hJT%2xuJGB*xI39%>TK;z>eN*r#RD$>Yl^LuJvBCJk&df+!Nb+K;%|}v{cew$E
zRBDEm|NJh0R%;IpDDgk;QKn49gSxcsRMK0EKYU)Y=w==a9jqB%^sKhH4P`E`>DKp3
zt(_akRxf?Y=l;(8h%td!l?V?MD?(2G0I1>bxQ~qH(@NN(G?*#|g;l?>Q1cNw&}t$*
zud)oYo#T&6St5HyS#utf`C+2oMK3qc%Y8my2+Qw(T?*BJEpH9GyCUpp6I5E{BVW&S
z^$#S31VTLgthr?-<j-?yJxNqIL&3lTh!WL12=f4l*eoO9eQ*hMhG2Bp&lXf!BF|Db
z=^4)mzIF^xkegf)qLPZ=i3VxpTuu3ap0_?OZY+YK=Qd$3aH{vgR>HUvj%n2?N%fip
z22bsO2P-*E5=L-7vb$-T>7a+QV(YRU$72)ddd~Xx^e!@`9TsHCjEqEM6CzL;L%kp8
zZ=iw^Gsy?vzfbVfRH>kwZOMOj@njgt>M_kL+(^MmmoY0VWDV%0Zq4#e*zyDlxjh`$
zKT6|Rrk}i@9#{^_7)W`8+29i=LGa|DaGBnJ-_{5e(b}X{G0jJrTCZ@mseiZJFFLV8
z^4BgH#^#9Zp)uE=!^p@UHr3Z$=dS?c$Z=o1&~)NN1$a&eLzj`OS*rzQii}YL+-s{q
z^5xjb^6#jEVYz=6T)^k=e>zLecVtkzeOwHe`0<Z3rt!;Wk^op=SaJQ_^e`QG=?qkV
ze6xgmJ1!~XHrvXCi7aZDvr6(K#jJ22e}^@6Lrh=%WcSo6k`7&9Qg*@s^jf`4H!Imj
z>1I%jRg8jMuQ<|A?nwwAoV4AVIi}JMEnn4<(f4H;+iAsO%g(No_W`aBaBsaCep-Br
z5h_Fj=L}UNbDBO?wuYE!qKELrD0y3d^GE8IWRx8fyJE^0Y0&HPscG5~Y9V3r4K+C1
zrAqY)6x|aP2Y|3X)n*sHU8E>Y`JqoYFm7=wa!SLwUGq&Vi+Cj(Im#)8AOubp=wg}U
z#9a43J8oT$cSxhvbtR0?g8V-;Pw7EWQk9f77mrIk+AO)utY4Od)~2!4WBXKpH;g92
zRCPIliY{J|fNB>JqMoK^YPz5`=+{)11@r4duOi0|OK>jr!hKpbT3SESdr5jUfHFlX
zGZr_DR<)-;u&N>JKhTuD4dLZd*Cv|nFB3r@k98!Uq8UK9QFi;%@3K()n0*CD^j3)s
z2GV-G!vF&3Qq2vEh-=8PF>RuMhlPV6#xR8kD!L&O>5-yO=sV}LIvUfW@wKYF%24Qr
zWex=NGBwJ)-D<xS^JZp{NYg53a_bCS-81fOcs!n7T7zPXyt@*WFJh)8iz;;}R<b+<
zz|GI95E82hV~qzPFCG~yGO0b-8EzGm+LW5t9kDYiDaCAn4E*=^XIc?v#u?DIKfKjl
z{$>9<WL&A(@!h<@d-k3zrDsLd!R_BSuyYJz8Q{_?^mu+lI*2@t0000100mesH842<
z00W3Hf&+vwf&y%y0|Eg80to<)`T5ciR%pyH4F(A+hDe6@4FLfQ1pt%12y!4G7Y#Bq
zFfuVSFf}zYGcsBj4>B<_GcYkSFfcVWGBYxhya<zjw`{?83~-cDPW?Od4tw*9G(qD2
zmyb}1beo`+CG)t8NRt~g>UC#PsiE&mlO^nR(wgqEx96~CtL>-Qem+;Ix*p8w8MgB&
zNwzGVYA&zfE(rxUrj5dJSBK`Gn2CHD3%*5%6hWswu+5)MrVJ^zoGN_VA_nT3zHBii
zS;RqqktHh7n$)pi?T&X>ad;sCfb(}=vC})OpKc&Q?S)|d+X0Q(MWkz|{z%*s&YF#u
z8?P8f-J7F1zudaAc_P2nM#KPFvZ*6jT24jokBc&nI|?LW9jTH-F3Jq$6K?4kvWR4(
zf6U&WL|JytxAG3pe?cGgEf^N?+^8BEsqs30jsF4z0RRD`Aut~>9R>qc9S#H*1Qdo)
zB|VqD-fUq(dX^^_;_Eq~D&Q~;1_>&LNQU<f0RamI00V*n0RW))g@KNsFCuATxU+M-
z?6|cTyoY`!8dEb=gDult_*9a%S<`YcBK|jgo!y6O?uG*Yb%f9>E3c6-j6obW$QaFk
z>GLsw9ZVv*B=W^no{B$gaWT74lRcm%?0YKqybhu#pqd((XQaUx`Mwr?XCjf0aWD|x
z-NWe3792mzK@p1;Mn{3}?in;Mw#jlSbz)z_Jdxl-hT4wk7|OI)#~v0@g_d<;DB88O
z!M$KPc=iEy#gXUI&;}Mm6J1tOd)`G<2wV9T1xo8iN!@YSd9etL9q586lLXR%*MS^-
z+>|<nx)p-sX;R~hznD6o^VJ<pnMN>m-4mxty)j<JTrzpwcx|PKMst;2c0S(*pT#?K
Hv}*zw@Pneu

literal 3726
zcmeH~`9IX#AIE2<nX!*8d$NW2ScWVa3`!`3VH8<nxRzp+Ez6K4jWvm~lti*rWJ}gj
z;^I=$h!9z_?_11wy5H~je((19{ss4k^Ei+5I_JC}=e*zNocHtW&+pHJKp@C(g~`Rq
z&)1g(0)Yef@attDxUB*P&;Ws;%m5h*1<2spJSYSVfpDWi+afjvLcFyrp92oDbGc(k
z-%$W76GArTWXvWj7>t6008S-#h6J^SL0On^dX@~kfCwL}vJ#-GtOBS3$CQ-`e5@)z
zno9p^{@oNZSp4^f!Eg|i3}y#0)`EnP!C+8bk^Budj`wYgzTzAexCZVbkttm9LB7-~
zJ$f|fr0MhcL>2eC+~W#j7hI_7jjGcv`2Onz+dIHg&Yltzbjkkbsj*`|GY0~vQUr&!
zeLxz1Y?E-)hK=ju=cl%}d01{ERU6Bb;p!RKvg|gzz8&i&iKYbwE>fotE`IMRV1L3|
zR$@YzQ#%jkQ%{Na>TZXLv1I)M92X*~5Zdq0bY`t(+kAOX6~kd?dp3>Hq0F`dklBrR
z<fyg3!>ub~JsgEM?WyKvS5IPD?LK?->VPzlWThP1j7h0D)hcBcPZjr!B~U&%FP3re
z==Oaw-@UrzUDmR5NijCZ^fb}!seu**3I>7m&jJ>JFvHivY%pFJx73+@EG$+lAn@@F
zDRjr5v*m2%WkB>t_hA@6z#H-R1K58IK!d?_7#x5ys^7bv5l@EO!622G1a{hF$pOQ6
z{LNihQ^MvX>v-6vf{tcf9B`T2_KTl7?Qlh|dV4u)HngN|!d!6LwXWoaK)!GHQH`Xa
ziQ$v!U8LM_RCqZRI5tYam8JlY){7^*+hyHtcm*ZXLlP$O_W~-zj6{}Soo{jI<C1H>
zf3WxLv1NEee3sQ*Vq^TkV#u6Hon6Tj&vUg0?QEvBT>=}XZnAl>Y(CUp$<p1tqMp~6
zlZEiBt*~^3_j4U-XD05ttkoyc#t7khU;G3gLM-v#pOA0qLtN5Oz7WMo?{`IyPRRrz
ziKziBQumAp;BS)Va=g6*%0$h2+L-r~IQ-|s8+vo|`MS&zkz<UT0tO*nygV-YkOKTK
z`u+Q+?5Tr5a4<lI?E!2|NI51j3@i)b1T)Smj0~I6g&B7zeU>r5aL>}hhFfDbA}D-K
zJkH`uZQr-JyOHF_)e9_zS?<kU>3Z#|U30Wa`}uG*;QdwlMOf<iC<mg!IstyA<PGI|
z%(_wibg?oaK~}BrXfT@vX<xt7L^?=d)VwuuqADD}OTcki9ga?xP`f6n^j0RW+vDA`
z1LdxFrEp@w!fxy(ExSkN(-kC}Vdo(0B#YUaK}>f?hN-SzyV3T6@r&VaBvO+i%QE#<
zb&U@0^Hd>qcnB&zk=clF8!drnHv?0uQBZ?hW2>*IVJ*^eHJabjUu8#_yMDG@gAlm;
z19%jPIb2OyHWBxk$~;P(Z|~~ca*;HaIWoi81}FnK1U{?nn|%gtGWe*xq2<w4ZNZM5
zVw%cREQBxW^z&28yRS8xFn5KEq7Z`7?K}4u=<bIM((b}C{Hi9#lgdUDKg45YJE`i<
zK902GShhx9D)MdOErBkQ{So|3e*l*G<-?tlM;kocT6ppy?P{U4g<`9D%h$4Lm1dHU
zPn6?(*3>6vA<~!%+X#MQ*{d@*D5u+9iuP&4gRuOTaR`sz;_UP%RCNcyTAOzJpgw<u
zLUJ`YcrY{Fq^@hsDX(TPCH$!3hPPsR#7O%MVHsl<deF<gD7qzoq-t$12`6>Mq+?M0
ze43$7^+0h*DF2&$l%l??{V)Z=k`wn!u)1aJBi3|)uH>vVidM+HPFYYt-Jg0HN4vAv
zExS11pLVW9&TOG+^-e~e<;NA@>RddKwo>6n6xiq~B_6U*OsCoUmZFze_Y^*FkG|)H
z0}z{y#?ud(1Yy@cesEci8L@26-!6_bOMmN@T{D_jSr?xYUAKaXp7$Cnj69CcC~OEL
z&U;z<i4bOn<neq<d=D;{-yh-!y|ZvKjx2SOXDdNiGn;3f@NnypZEw3Sh!G1p44IOd
zl>S1e4mHb;tb1f)@zEjim3N*^iAvm~1f2T*5xN`k%_{MW*Rjs|@gCF}g3cXOIk>_U
zC<xz>QuAvX(dn8tHAW}exeNxnXuUJLXR0!kRNN~aSTEnSAw4YBDlTFvU{2F&<*-$|
zz*}OcDJM7h@&*-m`r)Qr$uB(}G%1nDGr@usv-o3Al56DeJWJdKYm&R%9LKLAFO;9F
ztqk<gZ^~2*9<%HIDH}DB23s|*<lL*;`q|hj4@*31A2E<{hqfQ&@g!Db9e1l=QmN~e
zoJGU_Vd~pTFO3;5&%Cu!{n->`pWt&z4|n#Uii(txqT0~heJ`<L$MOF5U6zo!!|q%N
zq1xp^%Kmx1%Fu#^I)T{mS;w+?qpAsIX5Fc+2k@D6D{jSIKw;PL$UfykZ)JUbP1+#b
zU+-ak^V>e;YYb7JHkp#tGT@S!pDA!E#T7cduD~7+;_dW-zlk!CC`|%Q-hMzeq?OnF
z9GAFS5z&L0H+(6W^I|)br0J<Pyr+Q2bW(1wRB#P{Z53YmyoFsutf9`Q>|d*kteXv%
z#}%n5&$MEN?b>*;Mvu@cfR(Cv&uoH!^5WWECksw3mRFlG)dx8b#R=^mYNi*^dD3$1
zH2ij;XSrhe4p=HYmy;P)RLFNq-ddH}+o{8lM?PeTgcT14NL-E?@Pup#rEa}b&FJtr
zYagf<h>MlxYdx9s+(5QJOePJpO*i?jJTAQyYCGcW<9dtS*qmP3obfA0{Durj1~M=)
z()0oZZPJ)zP<X{;Db}z>S#m2|<`$;vPs!!)GXEc80rN*#zz_gmB@V`O%is$pG_$qO
zPd>g^eg3N)_>02)7X|qj1@%7#4RHRj3`PI$t!M1z|D(uDVhl49PZigxUKhDxGj_e8
z>4eJ#1LZMZK1n%tfl#-RDY#r>p%{|%x<mc@7mv3-Leys9n;j-M?JF(yfJj}Be5O^)
zy_(@{$+7Q74#EWw47H!#%$;%bTTxmO)<MvM>g98zq6A(fr2Bt4m-99H%tryWSc}B4
zokjhYl+Y)<At%q?rC%zkIGg<Bv`KH#i4ZC4lYqDST*6R6f`Z_}`$%%rf!Wd<r3d$9
zE?u3$+}E_$$P*juL0A|i%h~LLC~N)OB%?ZRM6(=AJf$!+i5cZ6y7PYP=Fxfetoe-1
zt9QK%*$es88yqHqhBwNK)e!%S$=MXv4$~Xx#cR)3Onya?KLz$zY5)VOsOcf`35m;N
zkJ`SH&HA?0R(|rOlS`Mcs*)vxqziXCL}xsM446h@HIF-<Z%1q#_i*j>cUQ}B$7M;s
z@MkjvsrJ-beZ~6Pik>iHNf13;#f&~S>%+cnH8;1T)HAWalteEqC}y5szNHVg;;D47
zMGBj^5ZALJVf_5BU5Qyo6o~7Mgqcs&P{)cv<LPM!XV%mZfh}i8_Mx&RwxXMcC7!my
zZDRe=H`P${7q2hU!TfH<yoJoQLYplaDv?uO4|5`-QXHY2bkC;pF?ip2FJh?-s~@MU
zZJQe{|BR<v#qQhK*C?lR1*g7;<Qo-hR&V;1yK9}do6aCEiTuOS|8Vqg|Du05n!(Ye
Yw(#Km+l!AMJzd*85>4gwWXcoy7YoouT>t<8

diff --git a/pulsar-broker/src/test/resources/authentication/keystoretls/client.truststore.jks b/pulsar-broker/src/test/resources/authentication/keystoretls/client.truststore.jks
index 210e423145de96274f254c6a8d48304cb63d9d03..ac59bd92541c5fdd5d11138ffa0bfd5278e9d92d 100644
GIT binary patch
literal 971
zcmezO_TO6u1_mY|W(3n*B}JvhCB-HAMX5lcHqTHNeg@VEJyQcq1_tI1gC^z{gC?e&
z1<Xv0OiY{%eLsI*5(vvUW5CPCsnzDu_MMlJk(-slAm32HfRBwil!Z;0Jv1*nFTXqw
zCc=Rs!j2)rg(1R?5D_*I1ewe%%$1yznVMHpTAW&Bpdij`WNct$Xl!6=YGiB_CBbiG
z2ox|hGJp!u-$_l3O31;&$jZRn#K_NJ(8S2a)Wpchus!EsIZwfqK)=7%U-+s&cbhmq
z`Zv8lptEH5f~l&XcXWG9mNtG>njW}v;d`&is&7g!&3d<S`|}NHtKTlaQD+yvVwc>R
zmy+8)YkF?enVY4z{(&ARtNF6tLxth(&*snQtdZp2=h`mfxZHNb+4;W9cr>@o(W<$v
z#P({|z8pi<$RmyuRkbe6y0S6hZC`nKVTB@N!{_q2jhC(0%+FJBeA}Av_cmki4c8^v
z%l~=Y5jr=kcbfEi3Aekm7hCSXvujhO(*CP%M;IbEtyGSP@^gLH-)+=q&83nkyK<7V
z-YK4^!g((xHg%*du0L~czDs1;+3lbB&euE2f6<i?`*3H4l;p~fR=xk3m>C%u7b_ab
z8^{9VQ<jfKj76j^K-G5o-g`NTj#bmt#UH)4T&VQ`IS_$q3mAxu3=6)uHuf#hSISJ<
zvAKB9n;l!l_q5lkNd+5+H0xfD{Sh*0Tjb>;L#2P_HFNK_XT57<`d`{|L0fzMM1vkj
z8M6}-XJ39cY>@R)+NtvCXvn-S`<z0<-2szr7pT6e*7~-GZ;|?fSyD67mmHM%xlgP%
zU1?%}p@G1?yN6$#6_c?)?I_qS>gL|~?wzEG{<f1vnx#qchioQ3aBjQR_d?>-mhj_p
zVu7vGN)t71ZQXKkZ-QmTH^#D~6Q5tYz$WG_92*u`eb1F6`ll$X*K0S=yM;F@H*)mI
zzG%{z%yOymdZSFuoheqWyF{BFX9hj)-ao@?{^zT*J~Q16O799U_uOk3cQnSR^3Kl>
YlQw4@<;px2Gp$_wbjZ!(2V3p{0B8Pf1poj5

literal 838
zcmezO_TO6u1_mY|W(3o0$%#ez`6WPZ#ozERfefq>dZq@J3=GWd22IS&22G5w7cet1
zGBJq=GyIebTPh>A^V7XjMSh`l(~>{j2E1&XT5TR}-+37sxmg(u1P%FtLd>Bo%sgz)
zjzK_k4dldmjSLM;jEoJ;3@nU{qQrTPL0m&97n_zQMkQo}8Ce;an;7{SfNtSpYGPz$
z*tJ5Zi%;On(f9dE0-K%piL8)&?z~D;e9|_@|1E;gd^UCV8K>`>Z>y(}kbJ~^zsZY3
z;a6LuB4!wTnD^I^jrl|Lndc8J@?LUOJfA3i*Cvm_vWV{)tIxh~txB=afBqEZnZ{+Z
zfBghj^Xc~M<{k}qisUU&Iq*(TsN{5>bLZa;2GQ=JS}IQ*1Z@^@=k4k3%sF?0MS*A5
z1p`}|662^73tyg|^<{3@qbo-gob6wq|L!SV!yX~U^!i&T*MrdFxG7}{=LMGcL>~!Q
zSMFiY8+kADq8)>k=B$Z|-`gf`_ByQQ+j&GOt39gsTGIP<0-_F=ZU_7>|B$=x(60jh
z_IW;j8EG3_teKb@85kD_8w4820>fICk420{M9qJ(JxjZFX~oKyB{jc_1rG&pDKwA=
zNh`BR7>G4sSHKTaAk4`4pM}+c8Au@qJ1{<h!OqBF+}6#1^Vu2>*UJ(I&)WM$eSgOL
zxbC~2omEGNL7_;*qayR0%A4n#pIFcRx@OJMCjrtgQg*G`Dz!NOoUUbm<&(P}Q_q&n
zujj5`f5gDzL6q~_i3UuEQ$5a|&`J*zlUAKt-Tf?lPU)69H@S~{V-Lk$64E}nQ1W81
z#YfhCowGvT^zHAw{=WK+_pZn_tFvNuN=Ak~w@I$p_q>NMljr*qn@_VGzL%LVxHNAT
zXVK2hK`E?Pg)~pFXZ%k7vZw#%!>D@4M@7<0n1aG{Z*!gee@952S-e`mG57L{WnLPU
zTp5!}dDP~3US~bt|7KopZs|IC-}6V=|Md$LzpdYQasFcQvjLoq4>K12=DOu_H_kI;
P&Afz-AH%LKe`W&!0!vD_

diff --git a/pulsar-broker/src/test/resources/authentication/keystoretls/client.truststore.nopassword.jks b/pulsar-broker/src/test/resources/authentication/keystoretls/client.truststore.nopassword.jks
new file mode 100644
index 0000000000000000000000000000000000000000..363fafab6be7a54925a2aefdd108509b785a5f80
GIT binary patch
literal 971
zcmezO_TO6u1_mY|W(3n*B}JvhCB-HAMX5lcHqTHNeg@VEJyQcq1_tI1gC^z{gC?e&
z1<Xv0OiY{%eLsI*5(vvUW5CPCsnzDu_MMlJk(-slAm32HfRBwil!Z;0Jv1*nFTXqw
zCc=Rs!j2)rg(1R?5D_*I1ewe%%$1yznVMHpTAW&Bpdij`WNct$Xl!6=YGiB_CBbiG
z2ox|hGJp!u-$_l3O31;&$jZRn#K_NJ(8S2a)Wpchus!EsIZwfqK)=7%U-+s&cbhmq
z`Zv8lptEH5f~l&XcXWG9mNtG>njW}v;d`&is&7g!&3d<S`|}NHtKTlaQD+yvVwc>R
zmy+8)YkF?enVY4z{(&ARtNF6tLxth(&*snQtdZp2=h`mfxZHNb+4;W9cr>@o(W<$v
z#P({|z8pi<$RmyuRkbe6y0S6hZC`nKVTB@N!{_q2jhC(0%+FJBeA}Av_cmki4c8^v
z%l~=Y5jr=kcbfEi3Aekm7hCSXvujhO(*CP%M;IbEtyGSP@^gLH-)+=q&83nkyK<7V
z-YK4^!g((xHg%*du0L~czDs1;+3lbB&euE2f6<i?`*3H4l;p~fR=xk3m>C%u7b_ab
z8^{9VQ<jfKj76j^K-G5o-g`NTj#bmt#UH)4T&VQ`IS_$q3mAxu3=6)uHuf#hSISJ<
zvAKB9n;l!l_q5lkNd+5+H0xfD{Sh*0Tjb>;L#2P_HFNK_XT57<`d`{|L0fzMM1vkj
z8M6}-XJ39cY>@R)+NtvCXvn-S`<z0<-2szr7pT6e*7~-GZ;|?fSyD67mmHM%xlgP%
zU1?%}p@G1?yN6$#6_c?)?I_qS>gL|~?wzEG{<f1vnx#qchioQ3aBjQR_d?>-mhj_p
zVu7vGN)t71ZQXKkZ-QmTH^#D~6Q5tYz$WG_92*u`eb1F6`ll$X*K0S=yM;F@H*)mI
zzG%{z%yOymdZSFuoheqWyF{BFX9hj)-ao@?{^zT*J~Q16O799U_uOk3cQnSR^3HZ{
X729dnmzGcD<c{xgaFWPYSKSBzHn?aJ

literal 0
HcmV?d00001

diff --git a/pulsar-broker/src/test/resources/authentication/keystoretls/proxy-and-client.truststore.jks b/pulsar-broker/src/test/resources/authentication/keystoretls/proxy-and-client.truststore.jks
new file mode 100644
index 0000000000000000000000000000000000000000..45a49018d8da3f7abb1b2afbd7e99bcdccbd3ba9
GIT binary patch
literal 1891
zcmezO_TO6u1_mZLW=qb=OwB7{U|?+X3}w5=z#5@vYGBF0z}#Wb#N1-g#FVpunTe5!
ziIbu4=g&(5VHsx(c-c6$+C196^D;7WvoaXu8wwcku`!3TunDt==4I#Qm*>GmI50%m
zF+{j9M7R+m!UlpMlbMCNAkHi;PAxJ}5a%^AHZU?YHZV0cGB%2m;5RY^3K$w0Kn3XU
zq$Wls<X~ZBWngY%<YzEwV&r0KVq|34o^!CAr(jB;-(Tx5eAS=3O&lNno8BMLSu%UU
zRMpQrx;-XK8^0<|4_vwMz1L*bH>H<mz1z6``G&OBZ<pVwvkPCbOYY1|$!(uCJ-6x1
z&C*-{K#!Bvd|B_I!tnNI^JjF{NOJFUZ5MG|ZoA>^eBWg}n%m}R)!bHMdo^ocj-hJg
z5yy$DS{G(r*_iOQuROf4LXolIb9vmx%hqe==P5Y8ZB6)lo3ZzX>yqr{|2*ypotxD=
zO?thA+uhlVE%)EqwW(5R|5djm43V2wDn~^5xxVY~HtMtHQc09uImub?6wg!Pyq6N2
zI#L$bpSd^RC9>@7_D_80>mB94=t_uvxU)h^a^**>-v3O@j0}v66%FJKWP$N1%f}+d
zBGMM1YCC=Jy_`hHs%h%tk6v3Y)Ovs%h`_W33`9nT1>aj6`xfXcWhU*|T)gMaj;-Q*
z+UwM$f{jC(buY*M2${4k@^X=((m(T>xp&*M-nB9PFKxM?t-XGtL64)1*$IiWFFzYL
z$oeSlRQYr?WL}qjPNCuMfXTKCRNqu<ecQvgNPWR9sTt`@4odvoCsv!TG_k+XK;Yip
z!!OQ?$=IKE6zmptb8me2PSQkw+sPu$(xmu9HWME>x83S{A#rL;_;ES0z}9J{i5j=I
zZaKI&!Ls5TW7*M(&o5nI6LS`h4GXNk=gJZNQ<T-~wVUVN!W)$vIeKJYG-*s`xzu>Q
zQKsh36sy)<qD_x8gC2M9pJ6rs^Ho`&nQjK9cZHXG?lp`%8e>#>2b@(|3ySh9Dxta5
z0Fg`UfVs2^m`jsjxpd;9&>3p}7v#`F(jW_#CuxvN`3!g<xfJ3`v<zyDo<WVH=;|6m
zd2`y!13iHYGMZgm?c%oCH~-qU=Wak^-8BWCZ~waQS})?|%06+fvm@H)*aDNPg#AsP
zKfd4kdqeG&f27e#-kkK$hfSW;Gw)4MSReeo<AB`CyoGyjv~Wf6oqxOj@zomPmMw{X
zjNkX#F64jxna44C&WvvFKKo1aRxsw~r93*bjOX}8`9zk9b0QLVFwB~#)_7&vQ=iU-
zZ|BatrNmWRFD`E3@v`|;vc}G%sp@~+@7iRXJaBA*i-=p)>i|Q=9WQQ*f4Sx|CFR)D
zB@-j3D^C7<*KYBYxFceVu2japTzf&BXZEDd=Y{)~AI)L4^UGO#Ez$6Xy=Awcu&S)V
zQdI-bOouC>mvaBgW6PV{`tF}BD7q@1Hb?ZsQsdH9>L&x1qNONs-ZU#%Zd&QUu|oWF
zQ_b}&OI|ogdRy6cUS4@bI3%9)LEN!Lu@jGfmh+CAohE81uDr`%Iq8g6gs4dB-)Ghu
zM|d7Ab?ds#a4)L)x=!L#hy4z-p6I^FEZ@D3y-iq8G(+yX|CjY<)fHY3?(*Gq4>)zB
zd$ra)CRWEc7RIyI1jJ5s-OoBp-BytE`XtS)6CV~OGADb@OFS0nocj1PtKp)VmTxWw
z20Z1S@li_U4$F)9^L~~Mw~oFr=d+l-W`P;|-pxl(i7oxh|L#u>|L1L<lJT`m723;=
zGp$J9dn|JA&&ZNpCHrTs=Bzs>5xbu)ugk6ON3z59Vue@VB{e0#td6<az^cA|>vgrO
Zx8`3syLB6TaPH})I}R`O2y{Br4**4<!A1Z8

literal 0
HcmV?d00001

diff --git a/pulsar-broker/src/test/resources/authentication/keystoretls/proxy.keystore.jks b/pulsar-broker/src/test/resources/authentication/keystoretls/proxy.keystore.jks
new file mode 100644
index 0000000000000000000000000000000000000000..d3b977560f190f103c267147ae21381f6ea31f22
GIT binary patch
literal 2245
zcmchYc{J1u8^`B2W-yGAEMuKUP1YO3SVDwCmKcgGp|K2EhD4U6%tT2kQnt8^y|Gnt
zBaDiXEm^W}mkeWTnYo6Jaj*Bi?|a^RPXE7ueE)bpe>~?o-{(2sXLo&f9Rvb__XPY~
z(C~<`nAp84+1eS<2Z5jfmJH!x5&S~x`~U<{1B(Fw6a*ndSYz!^D;aNelr_rp%dOSF
z`;6uq|G7elU#q^<^x+2}vbZ)$p7pI7QcmhxPv}%#`@S})krmXJKa02+L1^99#kKiF
zz-4Gr5ig%c9CN7`i0$5Sf;f@tgfVlO>pT3t=!`U?u?x?mWS&>(UZ00_c5;v*YzK(A
z1vuC5AM2?pDWCn`(th){4<u=I#E;m8+MQ!7v_5*QwbYT%zQ^yO8&sTX#K~;lF^~1D
z&c<KMQ1Mk&>rHD9w21Qb!p)($<Nb8@>c-YwYe!CSp4GeAiszcP?Tta_w%1;KdK}R9
zJGzH+w#oxPUpSvnd>5H#&$BR*z-DGYARaP652$EF0sBg<UpEbL!)Hd!vxIMdBGuq2
z^IDrXHZ*N)Beu}$lUQAkr<arO{~03%>AEAUS9HG#E@5i@B+M4J$k!ia9?V=C)llk7
z4ln5L<LHo<dyBb~D}vgRMU{!o9FO}2sEhlz><;!2L;TYz+ttv@z|9g)ecDpsI@yw8
zXs)B|?b3f)csZwyU?Ol=(0oOqES!ToO@9$-$XvQmV^x=5K=l=A*E#JTU``v2W?U^I
z?<)Qp3SU~Zt9R*ZQR$-f{0M9}i$WAjy5i_`%x#Kik*BD-Cr@f888r03C6^GOc-<+Q
zdRLq-<=;w~M?xF3!UB$FDGxRp|BzK2g5z6)K9$GR2ab_gHh!C|fk$ZulhyYcF5`y!
zsdwM1o;c#nsS#O>)R+;Il=GoUvd;dY(zYT@nJ9ua8=1udhvD)1jkg}TmQB0z$JZo7
zDARR>O;0bFN2z{uzq><7TtzK29PLu4!$jrES29BlRm9!e)H$bo`&nw31-G0yHf9*p
zL|x>n#dob$N%SRE*?hb@&0wRB$&zanXzzDk=_a)qVcMR{BzjOIQ<NB*=>Zj%dVf@V
zq9>}drj#B(TRkjvCSSU)TcRN<le!dnhUYLm@+j)dA?F*37MI#V`on`x=C?wM>6?|c
zZ$%RbXP#x8^!+8yktFXb>({@#PjJ%D=TF{{MYg<x;@5jAn#9cO^W*FfQ|B<PDq(LX
z=6q(h&*%5ee{`p+KfcD?vf$B9Lrus6FV>Ru9--|X_D*RP?;lq9j=O@<iCSib>ze7W
zim4Q}_HgidvS8s(E{!sitGVoY)V1T(oq@Csn@}{f$Kcya=81(VmZW@L!r(bevzhln
zF$Bx#4X!}Zx1DwL#;KS|88TM2fmBwg>y-S)&q&Pu0qKU-M|puGiPH8B@x#4evE?6p
zD(zhv;AhKE`FloCE9|NIBUf-?<{a(A$t|Z7B-!i8^tWMHr<q!Tur7Dh?D;EDeV9%x
zt3Nra?4p2^p?QS-%UqWe3QEuQ3F(?X?;mbYBHnp-q8`e$NNT9}hk6JKhJ<j4{8f5)
zFB6SUtegCf@pBT!Tp^~~H{Hy+q}J&;Wh~e;{@md#7^05&X5}F~{~5xY0RLJoUC}f1
z_TANkVw|SVW405_9++4WGtJIB``me>X#Z~u;$)^MBH-#5f*`|4Ox{1F!(=#)n<@=5
z3`Edh(h6bua=(4`X5{hN6VVcmC^`T63t2f{F!|s)IXXIntR{a9a_i)1mY+)5D1OX`
zA~*MX{Mzg^=ybDiL+OH?gGO(hV_s}y;_QH>)rC6F7yq=04834~J!IM6x#+5)i!}g&
z_!6*WzBnuy>{G}G2Ebq#D4pt@t>o|#g%#n4sU@FE=86D-FcgFh(%QEdn|#g?esKY3
zQZOkjn)F{H_!ANMi3t5fg#SknSops>^NIf**H|=CL>q_I*1}<Rb+vIgcWF^=tv#Tn
zjr|Y!zfMC24*t`Oy??<+21G$vG9Uyd0|4kTtL?61VIajK$;7MSIORt}^E952FoK3}
z?xs!~QALD;-+xHC?P13#)QP)5CtGiECwOB@D-LejT_Pa?>~}h!uk*EBK-Zq(-flxx
zl8Rc!ZVI^~`qyh0KE;dQtS8t5+!o^^(N#9w%=bxls%_fw!Ga1Pl;k(xTL$kOI6#1;
zKXE0z0_7GcB@UM@+NBh&=NC*I6uNdDi9C9WMd|icY3%S<-m#iK7TDFsC_E`)>As5B
zQhc>Mj^vD-%=BX{mZZC7DL&*)n-phywM$WlV?9~bACd69jFhEI^h5JcpeFVq)guJ0
zvEv4*V&aDtu%(Bw))!5Ood-jC2f%y)2q;#>9>6N>ZBYRMk%CAxq|J7PM|?sOpGbZ!
z#YH_;?!uS;)5N`t+A|RV>4lf;#+nLNAlc;j(czM1QyE)B<CLMwcJZ^`usJUV)ib@5
zjk5L1BT8x^4>dU)^66D`m6Y)3F^yE(;d7;ycP2qI?v&AA2#cn4)7;NTmM=!Xt`SHU
z*N_ZEjXH2@^{&R)%uOT4t?=DrsWi0$Fw|`AC@!}O@0o5vhvq69i@`=SRD<4sr4smj
zZ3+ks#}obwY^WAB$6#&15x*#${Y@4#1zGm)w>P*k(Xp(LIGR^gs3*|!qN7`?lqb5e
z6EDhcu$J+@R*JrLr4w8c(86$Q*>;O;ilpb#U<n_jJ?Z?UJC+GszNVv>(JNdTRT&QL
i{P+#1@^3FkmCk+N@SMCpqT;2;PwtGsG|-N-8vg?A72!Sr

literal 0
HcmV?d00001

diff --git a/pulsar-broker/src/test/resources/authentication/keystoretls/proxy.truststore.jks b/pulsar-broker/src/test/resources/authentication/keystoretls/proxy.truststore.jks
new file mode 100644
index 0000000000000000000000000000000000000000..0e13895b1c21d47158ef56c67a8e02b54fa79b5f
GIT binary patch
literal 971
zcmezO_TO6u1_mY|W(3n*B}JvhCB-HAMX5lcHqTJjN(R;lJyQcq1_tI1gC^z{gC?e&
z1<Xv0OiY{%eLsI*5(vvUW5CPCsnzDu_MMlJk(-slAm32HfRBwil!Z;0Jv1*nFTXqw
zCc=Rs!j2)rg(1R?5D_*I1ewe%%$1yznVMHpTAW&Bpdij`WNct$Xl!6=YGiB_CBbiG
z2ox|hGJp!u-$_l3O31;&$jZRn#K_NJ(8S2a)Wpchus!EsIZwfqK)=7%U-+s&cbhmq
z`Zv8lptEH5f~l&XcXWG9mNtG>njW}v;d`&is&7g!&3d<S`|}NHtKTlaQD+yvVwc>R
zmy+8)YkF?enVY4z{(&ARtNF6tLxth(&*snQtdZp2=h`mfxZHNb+4;W9cr>@o(W<$v
z#P({|z8pi<$RmyuRkbe6y0S6hZC`nKVTB@N!{_q2jhC(0%+FJBeA}Av_cmki4c8^v
z%l~=Y5jr=kcbfEi3Aekm7hCSXvujhO(*CP%M;IbEtyGSP@^gLH-)+=q&83nkyK<7V
z-YK4^!g((xHg%*du0L~czDs1;+3lbB&euE2f6<i?`*3H4l;p~fR=xk3m>C%u7b_ab
z8^{9VQ<jfKj76j^K-G5o-g`NTj#bmt#UH)4T&VQ`IS_$q3mAxu3=6)uHuf#hSISJ<
zvAKB9n;l!l_q5lkNd+5+H0xfD{Sh*0Tjb>;L#2P_HFNK_XT57<`d`{|L0fzMM1vkj
z8M6}-XJ39cY>@R)+NtvCXvn-S`<z0<-2szr7pT6e*7~-GZ;|?fSyD67mmHM%xlgP%
zU1?%}p@G1?yN6$#6_c?)?I_qS>gL|~?wzEG{<f1vnx#qchioQ3aBjQR_d?>-mhj_p
zVu7vGN)t71ZQXKkZ-QmTH^#D~6Q5tYz$WG_92*u`eb1F6`ll$X*K0S=yM;F@H*)mI
zzG%{z%yOymdZSFuoheqWyF{BFX9hj)-ao@?{^zT*J~Q16O799U_uOk3cQnSR@(!1e
XzeQq9k7D@MBHn+Q{(@?=OQrw-xX@_-

literal 0
HcmV?d00001

diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyAuthenticatedProducerConsumerTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyAuthenticatedProducerConsumerTest.java
index 622cc7dc35f59a..b6985400c81a16 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyAuthenticatedProducerConsumerTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyAuthenticatedProducerConsumerTest.java
@@ -23,6 +23,7 @@
 
 import com.google.common.collect.Sets;
 
+import java.util.Collections;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Optional;
@@ -57,9 +58,17 @@
 public class ProxyAuthenticatedProducerConsumerTest extends ProducerConsumerBase {
     private static final Logger log = LoggerFactory.getLogger(ProxyAuthenticatedProducerConsumerTest.class);
 
+    // Root for both proxy and client certificates
     private final String TLS_TRUST_CERT_FILE_PATH = "./src/test/resources/authentication/tls/cacert.pem";
-    private final String TLS_SERVER_CERT_FILE_PATH = "./src/test/resources/authentication/tls/server-cert.pem";
-    private final String TLS_SERVER_KEY_FILE_PATH = "./src/test/resources/authentication/tls/server-key.pem";
+
+    // Borrow certs for broker and proxy from other test
+    private final String TLS_PROXY_CERT_FILE_PATH = "./src/test/resources/authentication/tls/ProxyWithAuthorizationTest/proxy-cert.pem";
+    private final String TLS_PROXY_KEY_FILE_PATH = "./src/test/resources/authentication/tls/ProxyWithAuthorizationTest/proxy-key.pem";
+    private final String TLS_BROKER_TRUST_CERT_FILE_PATH = "./src/test/resources/authentication/tls/ProxyWithAuthorizationTest/broker-cacert.pem";
+    private final String TLS_BROKER_CERT_FILE_PATH = "./src/test/resources/authentication/tls/ProxyWithAuthorizationTest/broker-cert.pem";
+    private final String TLS_BROKER_KEY_FILE_PATH = "./src/test/resources/authentication/tls/ProxyWithAuthorizationTest/broker-key.pem";
+
+    // This client cert is a superUser, so use that one
     private final String TLS_CLIENT_CERT_FILE_PATH = "./src/test/resources/authentication/tls/client-cert.pem";
     private final String TLS_CLIENT_KEY_FILE_PATH = "./src/test/resources/authentication/tls/client-key.pem";
 
@@ -78,20 +87,23 @@ protected void setup() throws Exception {
         conf.setBrokerServicePortTls(Optional.of(0));
         conf.setWebServicePortTls(Optional.of(0));
         conf.setTlsTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH);
-        conf.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
-        conf.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
-        conf.setTlsAllowInsecureConnection(true);
+        conf.setTlsCertificateFilePath(TLS_BROKER_CERT_FILE_PATH);
+        conf.setTlsKeyFilePath(TLS_BROKER_KEY_FILE_PATH);
+        conf.setTlsAllowInsecureConnection(false);
         conf.setNumExecutorThreadPoolSize(5);
 
         Set<String> superUserRoles = new HashSet<>();
         superUserRoles.add("localhost");
         superUserRoles.add("superUser");
+        superUserRoles.add("Proxy");
         conf.setSuperUserRoles(superUserRoles);
+        conf.setProxyRoles(Collections.singleton("Proxy"));
 
         conf.setBrokerClientAuthenticationPlugin(AuthenticationTls.class.getName());
         conf.setBrokerClientAuthenticationParameters(
-                "tlsCertFile:" + TLS_CLIENT_CERT_FILE_PATH + "," + "tlsKeyFile:" + TLS_SERVER_KEY_FILE_PATH);
-        conf.setBrokerClientTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH);
+                "tlsCertFile:" + TLS_CLIENT_CERT_FILE_PATH + "," + "tlsKeyFile:" + TLS_CLIENT_KEY_FILE_PATH);
+        conf.setBrokerClientTrustCertsFilePath(TLS_BROKER_TRUST_CERT_FILE_PATH);
+        conf.setBrokerClientTlsEnabled(true);
         Set<String> providers = new HashSet<>();
         providers.add(AuthenticationProviderTls.class.getName());
         conf.setAuthenticationProviders(providers);
@@ -102,7 +114,6 @@ protected void setup() throws Exception {
 
         // start proxy service
         proxyConfig.setAuthenticationEnabled(true);
-        proxyConfig.setAuthenticationEnabled(true);
 
         proxyConfig.setServicePort(Optional.of(0));
         proxyConfig.setBrokerProxyAllowedTargetPorts("*");
@@ -110,16 +121,18 @@ protected void setup() throws Exception {
         proxyConfig.setWebServicePort(Optional.of(0));
         proxyConfig.setWebServicePortTls(Optional.of(0));
         proxyConfig.setTlsEnabledWithBroker(true);
+        // Setting advertised address to localhost to avoid hostname verification failure
+        proxyConfig.setAdvertisedAddress("localhost");
 
         // enable tls and auth&auth at proxy
-        proxyConfig.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
-        proxyConfig.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
+        proxyConfig.setTlsCertificateFilePath(TLS_PROXY_CERT_FILE_PATH);
+        proxyConfig.setTlsKeyFilePath(TLS_PROXY_KEY_FILE_PATH);
         proxyConfig.setTlsTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH);
 
         proxyConfig.setBrokerClientAuthenticationPlugin(AuthenticationTls.class.getName());
         proxyConfig.setBrokerClientAuthenticationParameters(
-                "tlsCertFile:" + TLS_CLIENT_CERT_FILE_PATH + "," + "tlsKeyFile:" + TLS_CLIENT_KEY_FILE_PATH);
-        proxyConfig.setBrokerClientTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH);
+                "tlsCertFile:" + TLS_PROXY_CERT_FILE_PATH + "," + "tlsKeyFile:" + TLS_PROXY_KEY_FILE_PATH);
+        proxyConfig.setBrokerClientTrustCertsFilePath(TLS_BROKER_TRUST_CERT_FILE_PATH);
         proxyConfig.setAuthenticationProviders(providers);
 
         proxyConfig.setMetadataStoreUrl(DUMMY_VALUE);
@@ -207,10 +220,11 @@ public void testTlsSyncProducerAndConsumer() throws Exception {
     }
 
     protected final PulsarClient createPulsarClient(Authentication auth, String lookupUrl) throws Exception {
-        admin = spy(PulsarAdmin.builder().serviceHttpUrl(brokerUrlTls.toString()).tlsTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH)
-                .allowTlsInsecureConnection(true).authentication(auth).build());
+        admin = spy(PulsarAdmin.builder().serviceHttpUrl(brokerUrlTls.toString())
+                .tlsTrustCertsFilePath(TLS_BROKER_TRUST_CERT_FILE_PATH)
+                .enableTlsHostnameVerification(true).authentication(auth).build());
         return PulsarClient.builder().serviceUrl(lookupUrl).statsInterval(0, TimeUnit.SECONDS)
-                .tlsTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH).allowTlsInsecureConnection(true).authentication(auth)
+                .tlsTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH).enableTlsHostnameVerification(true).authentication(auth)
                 .enableTls(true).build();
 
     }
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationNegTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationNegTest.java
index 0fad961ba2110d..23138929c84b50 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationNegTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationNegTest.java
@@ -22,6 +22,7 @@
 
 import com.google.common.collect.Sets;
 
+import java.util.Collections;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Optional;
@@ -91,6 +92,7 @@ protected void setup() throws Exception {
         Set<String> superUserRoles = new HashSet<>();
         superUserRoles.add("superUser");
         conf.setSuperUserRoles(superUserRoles);
+        conf.setProxyRoles(Collections.singleton("Proxy"));
 
         conf.setBrokerClientAuthenticationPlugin(AuthenticationTls.class.getName());
         conf.setBrokerClientAuthenticationParameters(
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithJwtAuthorizationTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithJwtAuthorizationTest.java
index 07b71530b53a6c..0775722d504e05 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithJwtAuthorizationTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithJwtAuthorizationTest.java
@@ -22,11 +22,15 @@
 
 import com.google.common.collect.Sets;
 
-import java.util.*;
-import java.util.concurrent.TimeUnit;
-
 import io.jsonwebtoken.Jwts;
 import io.jsonwebtoken.SignatureAlgorithm;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import javax.crypto.SecretKey;
 import org.apache.pulsar.broker.authentication.AuthenticationProviderToken;
 import org.apache.pulsar.broker.authentication.AuthenticationService;
 import org.apache.pulsar.broker.authentication.utils.AuthTokenUtils;
@@ -49,7 +53,7 @@
 import javax.crypto.SecretKey;
 
 public class ProxyWithJwtAuthorizationTest extends ProducerConsumerBase {
-    private static final Logger log = LoggerFactory.getLogger(ProxyWithAuthorizationTest.class);
+    private static final Logger log = LoggerFactory.getLogger(ProxyWithJwtAuthorizationTest.class);
 
     private final String ADMIN_ROLE = "admin";
     private final String PROXY_ROLE = "proxy";
@@ -78,6 +82,7 @@ protected void setup() throws Exception {
         superUserRoles.add(PROXY_ROLE);
         superUserRoles.add(BROKER_ROLE);
         conf.setSuperUserRoles(superUserRoles);
+        conf.setProxyRoles(Collections.singleton(PROXY_ROLE));
 
         conf.setBrokerClientAuthenticationPlugin(AuthenticationToken.class.getName());
         conf.setBrokerClientAuthenticationParameters(BROKER_TOKEN);