Separating configuration for client and server trust store (apache#1246

)

* Use brokerClientTlsTrustCertsFilePath to configure the trust file path for outgoing connection to a broker

* Separating configuration for client and server trust store

* Addressed Matteo's PR Comments

conf/broker.conf

@@ -234,6 +234,7 @@ superUserRoles=
 # either in same or other clusters
 brokerClientAuthenticationPlugin=
 brokerClientAuthenticationParameters=
+brokerClientTrustCertsFilePath=
 
 # Supported Athenz provider domain names(comma separated) for authentication
 athenzDomainNames=

conf/proxy.conf

@@ -59,6 +59,7 @@ authorizationProvider=org.apache.pulsar.broker.authorization.PulsarAuthorization
 # Authentication settings of the proxy itself. Used to connect to brokers
 brokerClientAuthenticationPlugin=
 brokerClientAuthenticationParameters=
+brokerClientTrustCertsFilePath=
 
 # Role names that are treated as "super-user", meaning they will be able to do all admin
 # operations and publish/consume from all topics (comma-separated)

conf/websocket.conf

@@ -78,6 +78,7 @@ superUserRoles=
 # Authentication settings of the proxy itself. Used to connect to brokers
 brokerClientAuthenticationPlugin=
 brokerClientAuthenticationParameters=
+brokerClientTrustCertsFilePath=
 
 # When this parameter is not empty, unauthenticated users perform as anonymousUserRole
 anonymousUserRole=

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -234,7 +234,9 @@ public class ServiceConfiguration implements PulsarConfiguration {
     // to other brokers, either in same or other clusters. Default uses plugin which disables authentication
     private String brokerClientAuthenticationPlugin = "org.apache.pulsar.client.impl.auth.AuthenticationDisabled";
     private String brokerClientAuthenticationParameters = "";
-
+    // Path for the trusted TLS certificate file for outgoing connection to a server (broker)
+    private String brokerClientTrustCertsFilePath = "";
+
     // When this parameter is not empty, unauthenticated users perform as anonymousUserRole
     private String anonymousUserRole = null;
 
@@ -894,6 +896,14 @@ public void setBrokerClientAuthenticationParameters(String brokerClientAuthentic
         this.brokerClientAuthenticationParameters = brokerClientAuthenticationParameters;
     }
 
+    public String getBrokerClientTrustCertsFilePath() {
+        return brokerClientTrustCertsFilePath;
+    }
+
+    public void setBrokerClientTrustCertsFilePath(String brokerClientTrustCertsFilePath) {
+        this.brokerClientTrustCertsFilePath = brokerClientTrustCertsFilePath;
+    }
+
     public String getAnonymousUserRole() {
         return anonymousUserRole;
     }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

@@ -512,7 +512,7 @@ public PulsarClient getReplicationClient(String cluster) {
                     clusterUrl = isNotBlank(data.getBrokerServiceUrlTls()) ? data.getBrokerServiceUrlTls()
                             : data.getServiceUrlTls();
                     configuration.setUseTls(true);
-                    configuration.setTlsTrustCertsFilePath(pulsar.getConfiguration().getTlsTrustCertsFilePath());
+                    configuration.setTlsTrustCertsFilePath(pulsar.getConfiguration().getBrokerClientTrustCertsFilePath());
                     configuration
                             .setTlsAllowInsecureConnection(pulsar.getConfiguration().isTlsAllowInsecureConnection());
                 } else {

pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPublishConsumeTlsTest.java

@@ -68,6 +68,7 @@ public void setup() throws Exception {
         config.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
         config.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         config.setTlsTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH);
+        config.setBrokerClientTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH);
         config.setClusterName("use");
         config.setGlobalZookeeperServers("dummy-zk-servers");
         service = spy(new WebSocketService(config));

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/DirectProxyHandler.java

@@ -87,11 +87,11 @@ protected void initChannel(SocketChannel ch) throws Exception {
                     AuthenticationDataProvider authData = authentication.getAuthData();
                     if (authData.hasDataForTls()) {
                         sslCtx = SecurityUtility.createNettySslContextForClient(config.isTlsAllowInsecureConnection(),
-                                config.getTlsTrustCertsFilePath(), (X509Certificate[]) authData.getTlsCertificates(),
+                                config.getBrokerClientTrustCertsFilePath(), (X509Certificate[]) authData.getTlsCertificates(),
                                 authData.getTlsPrivateKey());
                     } else {
                         sslCtx = SecurityUtility.createNettySslContextForClient(config.isTlsAllowInsecureConnection(),
-                                config.getTlsTrustCertsFilePath());
+                                config.getBrokerClientTrustCertsFilePath());
                     }
                     ch.pipeline().addLast(TLS_HANDLER, sslCtx.newHandler(ch.alloc()));
                 }

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConfiguration.java

@@ -35,11 +35,11 @@ public class ProxyConfiguration implements PulsarConfiguration {
 
     // ZooKeeper session timeout
     private int zookeeperSessionTimeoutMs = 30_000;
-    
-    // if Service Discovery is Disabled this url should point to the discovery service provider. 
+
+    // if Service Discovery is Disabled this url should point to the discovery service provider.
     private String brokerServiceURL;
     private String brokerServiceURLTLS;
-    
+
     // Port to use to server binary-proto request
     private int servicePort = 6650;
     // Port to use to server binary-proto-tls request
@@ -73,6 +73,7 @@ public class ProxyConfiguration implements PulsarConfiguration {
     // Authentication settings of the proxy itself. Used to connect to brokers
     private String brokerClientAuthenticationPlugin;
     private String brokerClientAuthenticationParameters;
+    private String brokerClientTrustCertsFilePath;
 
     /***** --- TLS --- ****/
     // Enable TLS for the proxy handler
@@ -97,33 +98,33 @@ public class ProxyConfiguration implements PulsarConfiguration {
     // Specify the tls cipher the broker will use to negotiate during TLS Handshake.
     // Example:- [TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256]
     private Set<String> tlsCiphers = Sets.newTreeSet();
-    
+
     private Properties properties = new Properties();
 
     public boolean forwardAuthorizationCredentials() {
         return forwardAuthorizationCredentials;
     }
-    
+
     public void setForwardAuthorizationCredentials(boolean forwardAuthorizationCredentials) {
         this.forwardAuthorizationCredentials = forwardAuthorizationCredentials;
     }
-    
+
     public String getBrokerServiceURLTLS() {
         return brokerServiceURLTLS;
     }
-    
+
     public void setBrokerServiceURLTLS(String discoveryServiceURLTLS) {
         this.brokerServiceURLTLS = discoveryServiceURLTLS;
     }
-    
+
     public String getBrokerServiceURL() {
         return brokerServiceURL;
     }
-    
+
     public void setBrokerServiceURL(String discoveryServiceURL) {
         this.brokerServiceURL = discoveryServiceURL;
     }
-    
+
     public String getZookeeperServers() {
         return zookeeperServers;
     }
@@ -260,6 +261,14 @@ public void setBrokerClientAuthenticationParameters(String brokerClientAuthentic
         this.brokerClientAuthenticationParameters = brokerClientAuthenticationParameters;
     }
 
+    public String getBrokerClientTrustCertsFilePath() {
+        return this.brokerClientTrustCertsFilePath;
+    }
+
+    public void setBrokerClientTrustCertsFilePath(String brokerClientTlsTrustCertsFilePath) {
+        this.brokerClientTrustCertsFilePath = brokerClientTlsTrustCertsFilePath;
+    }
+
     public boolean isAuthenticationEnabled() {
         return authenticationEnabled;
     }
@@ -307,7 +316,7 @@ public Properties getProperties() {
     public void setProperties(Properties properties) {
         this.properties = properties;
     }
-    
+
     public Set<String> getTlsProtocols() {
         return tlsProtocols;
     }

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyService.java

@@ -102,7 +102,7 @@ public ProxyService(ProxyConfiguration proxyConfig) throws IOException {
         }
         if (proxyConfig.isTlsEnabledWithBroker()) {
             clientConfiguration.setUseTls(true);
-            clientConfiguration.setTlsTrustCertsFilePath(proxyConfig.getTlsTrustCertsFilePath());
+            clientConfiguration.setTlsTrustCertsFilePath(proxyConfig.getBrokerClientTrustCertsFilePath());
             clientConfiguration.setTlsAllowInsecureConnection(proxyConfig.isTlsAllowInsecureConnection());
         }
 

pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyAuthenticatedProducerConsumerTest.java

@@ -89,7 +89,7 @@ protected void setup() throws Exception {
         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);
         Set<String> providers = new HashSet<>();
         providers.add(AuthenticationProviderTls.class.getName());
         conf.setAuthenticationProviders(providers);
@@ -113,10 +113,11 @@ protected void setup() throws Exception {
         proxyConfig.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
         proxyConfig.setTlsKeyFilePath(TLS_SERVER_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);
         proxyConfig.setAuthenticationProviders(providers);
 
         proxyConfig.setZookeeperServers(DUMMY_VALUE);

pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithProxyAuthorizationNegTest.java → pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/ProxyWithAuthorizationNegTest.java

@@ -54,18 +54,18 @@
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
 
-public class ProxyWithProxyAuthorizationNegTest extends ProducerConsumerBase {
-    private static final Logger log = LoggerFactory.getLogger(ProxyWithProxyAuthorizationNegTest.class);
-
-    private final String TLS_PROXY_TRUST_CERT_FILE_PATH = "./src/test/resources/authentication/tls/ProxyWithProxyAuthorizationTest/cacert.pem";
-    private final String TLS_PROXY_CERT_FILE_PATH = "./src/test/resources/authentication/tls/ProxyWithProxyAuthorizationTest/proxy-cert.pem";
-    private final String TLS_PROXY_KEY_FILE_PATH = "./src/test/resources/authentication/tls/ProxyWithProxyAuthorizationTest/proxy-key.pem";
-    private final String TLS_SERVER_CERT_TRUST_FILE_PATH = "./src/test/resources/authentication/tls/ProxyWithProxyAuthorizationTest/cacert.pem";
-    private final String TLS_SERVER_CERT_FILE_PATH = "./src/test/resources/authentication/tls/ProxyWithProxyAuthorizationTest/broker-cert.pem";
-    private final String TLS_SERVER_KEY_FILE_PATH = "./src/test/resources/authentication/tls/ProxyWithProxyAuthorizationTest/broker-key.pem";
-    private final String TLS_CLIENT_TRUST_CERT_FILE_PATH = "./src/test/resources/authentication/tls/ProxyWithProxyAuthorizationTest/cacert.pem";
-    private final String TLS_CLIENT_CERT_FILE_PATH = "./src/test/resources/authentication/tls/ProxyWithProxyAuthorizationTest/client-cert.pem";
-    private final String TLS_CLIENT_KEY_FILE_PATH = "./src/test/resources/authentication/tls/ProxyWithProxyAuthorizationTest/client-key.pem";
+public class ProxyWithAuthorizationNegTest extends ProducerConsumerBase {
+    private static final Logger log = LoggerFactory.getLogger(ProxyWithAuthorizationNegTest.class);
+
+    private final String TLS_PROXY_TRUST_CERT_FILE_PATH = "./src/test/resources/authentication/tls/ProxyWithAuthorizationTest/proxy-cacert.pem";
+    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";
+    private final String TLS_CLIENT_TRUST_CERT_FILE_PATH = "./src/test/resources/authentication/tls/ProxyWithAuthorizationTest/client-cacert.pem";
+    private final String TLS_CLIENT_CERT_FILE_PATH = "./src/test/resources/authentication/tls/ProxyWithAuthorizationTest/client-cert.pem";
+    private final String TLS_CLIENT_KEY_FILE_PATH = "./src/test/resources/authentication/tls/ProxyWithAuthorizationTest/client-key.pem";
     private final String TLS_SUPERUSER_CLIENT_KEY_FILE_PATH = "./src/test/resources/authentication/tls/client-key.pem";
     private final String TLS_SUPERUSER_CLIENT_CERT_FILE_PATH = "./src/test/resources/authentication/tls/client-cert.pem";
     private final String TLS_SUPERUSER_CLIENT_TRUST_CERT_FILE_PATH = "./src/test/resources/authentication/tls/cacert.pem";
@@ -82,9 +82,9 @@ protected void setup() throws Exception {
         conf.setAuthorizationEnabled(true);
 
         conf.setTlsEnabled(true);
-        conf.setTlsTrustCertsFilePath(TLS_SERVER_CERT_TRUST_FILE_PATH);
-        conf.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH);
-        conf.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH);
+        conf.setTlsTrustCertsFilePath(TLS_PROXY_TRUST_CERT_FILE_PATH);
+        conf.setTlsCertificateFilePath(TLS_BROKER_CERT_FILE_PATH);
+        conf.setTlsKeyFilePath(TLS_BROKER_KEY_FILE_PATH);
         conf.setTlsAllowInsecureConnection(true);
 
         Set<String> superUserRoles = new HashSet<>();
@@ -93,7 +93,7 @@ protected void setup() throws Exception {
 
         conf.setBrokerClientAuthenticationPlugin(AuthenticationTls.class.getName());
         conf.setBrokerClientAuthenticationParameters(
-                "tlsCertFile:" + TLS_SERVER_CERT_FILE_PATH + "," + "tlsKeyFile:" + TLS_SERVER_KEY_FILE_PATH);
+                "tlsCertFile:" + TLS_BROKER_CERT_FILE_PATH + "," + "tlsKeyFile:" + TLS_BROKER_KEY_FILE_PATH);
 
         Set<String> providers = new HashSet<>();
         providers.add(AuthenticationProviderTls.class.getName());
@@ -119,11 +119,13 @@ protected void setup() throws Exception {
         // enable tls and auth&auth at proxy
         proxyConfig.setTlsCertificateFilePath(TLS_PROXY_CERT_FILE_PATH);
         proxyConfig.setTlsKeyFilePath(TLS_PROXY_KEY_FILE_PATH);
-        proxyConfig.setTlsTrustCertsFilePath(TLS_PROXY_TRUST_CERT_FILE_PATH);
+        proxyConfig.setTlsTrustCertsFilePath(TLS_CLIENT_TRUST_CERT_FILE_PATH);
 
         proxyConfig.setBrokerClientAuthenticationPlugin(AuthenticationTls.class.getName());
         proxyConfig.setBrokerClientAuthenticationParameters(
                 "tlsCertFile:" + TLS_PROXY_CERT_FILE_PATH + "," + "tlsKeyFile:" + TLS_PROXY_KEY_FILE_PATH);
+        proxyConfig.setBrokerClientTrustCertsFilePath(TLS_BROKER_TRUST_CERT_FILE_PATH);
+
         proxyConfig.setAuthenticationProviders(providers);
 
         proxyService = Mockito.spy(new ProxyService(proxyConfig));
@@ -225,7 +227,7 @@ protected final void createAdminClient() throws Exception {
         authTls.configure(authParams);
         org.apache.pulsar.client.api.ClientConfiguration clientConf = new org.apache.pulsar.client.api.ClientConfiguration();
         clientConf.setStatsInterval(0, TimeUnit.SECONDS);
-        clientConf.setTlsTrustCertsFilePath(TLS_SUPERUSER_CLIENT_TRUST_CERT_FILE_PATH);
+        clientConf.setTlsTrustCertsFilePath(TLS_BROKER_TRUST_CERT_FILE_PATH);
         clientConf.setTlsAllowInsecureConnection(true);
         clientConf.setAuthentication(authTls);
         clientConf.setUseTls(true);
@@ -241,7 +243,7 @@ private PulsarClient createPulsarClient(String proxyServiceUrl) throws PulsarCli
         authTls.configure(authParams);
         org.apache.pulsar.client.api.ClientConfiguration clientConf = new org.apache.pulsar.client.api.ClientConfiguration();
         clientConf.setStatsInterval(0, TimeUnit.SECONDS);
-        clientConf.setTlsTrustCertsFilePath(TLS_CLIENT_TRUST_CERT_FILE_PATH);
+        clientConf.setTlsTrustCertsFilePath(TLS_PROXY_TRUST_CERT_FILE_PATH);
         clientConf.setTlsAllowInsecureConnection(true);
         clientConf.setAuthentication(authTls);
         clientConf.setUseTls(true);