ZOOKEEPER-3806: TLS - dynamic loading for client trust/key store

ZooKeer currently has support for reloading the Quorum Truststore & Keystore automatically when the certificate files change in the filesystem without server restart (https://github.com/apache/zookeeper/pull/680)

However, Reloading of key and trust store for **ClientX509Util** is not present; i.e., the server presented certs to the clients will not get reloaded automatically if the certificates in the filesystem change, short-lived certs requires the process restart.

Changes:

-  A new config property "zookeeper.client.certReload" is added, if it's true - ClientX509Util is reloaded automatically.

-  ZK uses an _X509AuthenticationProvider_ which is backed by an X509TrustManager and an X509KeyManager to perform _remote host certificate authentication_. We need to update the X509AuthenticationProvider's TrustStore as part of the X509Util file-watcher.
- Junit test case to verify the cert reload.

Author: Manu Mathew <manu.mathew@netapp.com>
Author: mathewmanu <manmathew@cs.stonybrook.edu>
Author: Manu Mathew <101424654+mathew-manu@users.noreply.github.com>

Reviewers: Enrico Olivelli <eolivelli@apache.org>, Mate Szalay-Beko <symat@apache.org>

Closes #1839 from mathew-manu/ZOOKEEPER-3806

5 files changed, 274 insertions, 10 deletions

diff --git a/zookeeper-docs/src/main/resources/markdown/zookeeperAdmin.md b/zookeeper-docs/src/main/resources/markdown/zookeeperAdmin.md
index 23e85804b..f7e38b926 100644
--- a/zookeeper-docs/src/main/resources/markdown/zookeeperAdmin.md
+++ b/zookeeper-docs/src/main/resources/markdown/zookeeperAdmin.md
@@ -1740,6 +1740,16 @@ and [SASL authentication for ZooKeeper](https://cwiki.apache.org/confluence/disp
     **New in 3.5.5:**
     TBD
 
+* *sslQuorumReloadCertFiles* :
+    (No Java system property)
+    **New in  3.5.5, 3.6.0:**
+    Allows Quorum SSL keyStore and trustStore reloading when the certificates on the filesystem change without having to restart the ZK process. Default: false
+
+* *client.certReload* :
+    (Java system property: **zookeeper.client.certReload**)
+    **New in 3.9.0:**
+    Allows client SSL keyStore and trustStore reloading when the certificates on the filesystem change without having to restart the ZK process. Default: false
+
 * *client.portUnification*:
     (Java system property: **zookeeper.client.portUnification**)
     Specifies that the client port should accept SSL connections
diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/common/X509Util.java b/zookeeper-server/src/main/java/org/apache/zookeeper/common/X509Util.java
index 0cc3eb218..1fbbc7205 100644
--- a/zookeeper-server/src/main/java/org/apache/zookeeper/common/X509Util.java
+++ b/zookeeper-server/src/main/java/org/apache/zookeeper/common/X509Util.java
@@ -50,6 +50,8 @@ import javax.net.ssl.X509TrustManager;
 import org.apache.zookeeper.common.X509Exception.KeyManagerException;
 import org.apache.zookeeper.common.X509Exception.SSLContextException;
 import org.apache.zookeeper.common.X509Exception.TrustManagerException;
+import org.apache.zookeeper.server.NettyServerCnxnFactory;
+import org.apache.zookeeper.server.auth.ProviderRegistry;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -281,6 +283,10 @@ public abstract class X509Util implements Closeable, AutoCloseable {
     private void resetDefaultSSLContextAndOptions() throws X509Exception.SSLContextException {
         SSLContextAndOptions newContext = createSSLContextAndOptions();
         defaultSSLContextAndOptions.set(newContext);
+
+        if (Boolean.getBoolean(NettyServerCnxnFactory.CLIENT_CERT_RELOAD_KEY)) {
+            ProviderRegistry.addOrUpdateProvider(ProviderRegistry.AUTHPROVIDER_PROPERTY_PREFIX + "x509");
+        }
     }
 
     private SSLContextAndOptions createSSLContextAndOptions() throws SSLContextException {
diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/NettyServerCnxnFactory.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/NettyServerCnxnFactory.java
index 040650e4a..68335485b 100644
--- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/NettyServerCnxnFactory.java
+++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/NettyServerCnxnFactory.java
@@ -124,6 +124,8 @@ public class NettyServerCnxnFactory extends ServerCnxnFactory {
 
     private static final AtomicReference<ByteBufAllocator> TEST_ALLOCATOR = new AtomicReference<>(null);
 
+    public static final String CLIENT_CERT_RELOAD_KEY = "zookeeper.client.certReload";
+
     /**
      * A handler that detects whether the client would like to use
      * TLS or not and responds in kind. The first bytes are examined
@@ -514,6 +516,17 @@ public class NettyServerCnxnFactory extends ServerCnxnFactory {
     NettyServerCnxnFactory() {
         x509Util = new ClientX509Util();
 
+        boolean useClientReload = Boolean.getBoolean(CLIENT_CERT_RELOAD_KEY);
+        LOG.info("{}={}", CLIENT_CERT_RELOAD_KEY, useClientReload);
+        if (useClientReload) {
+            try {
+                x509Util.enableCertFileReloading();
+            } catch (IOException e) {
+                LOG.error("unable to set up client certificate reload filewatcher", e);
+                useClientReload = false;
+            }
+        }
+
         boolean usePortUnification = Boolean.getBoolean(PORT_UNIFICATION_KEY);
 
         LOG.info("{}={}", PORT_UNIFICATION_KEY, usePortUnification);
diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/auth/ProviderRegistry.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/auth/ProviderRegistry.java
index 3315e44ed..bab8059e0 100644
--- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/auth/ProviderRegistry.java
+++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/auth/ProviderRegistry.java
@@ -55,21 +55,27 @@ public class ProviderRegistry {
             Enumeration<Object> en = System.getProperties().keys();
             while (en.hasMoreElements()) {
                 String k = (String) en.nextElement();
-                if (k.startsWith(AUTHPROVIDER_PROPERTY_PREFIX)) {
-                    String className = System.getProperty(k);
-                    try {
-                        Class<?> c = ZooKeeperServer.class.getClassLoader().loadClass(className);
-                        AuthenticationProvider ap = (AuthenticationProvider) c.getDeclaredConstructor().newInstance();
-                        authenticationProviders.put(ap.getScheme(), ap);
-                    } catch (Exception e) {
-                        LOG.warn("Problems loading {}", className, e);
-                    }
-                }
+                addOrUpdateProvider(k);
             }
             initialized = true;
         }
     }
 
+    public static void addOrUpdateProvider(String authKey) {
+        synchronized (ProviderRegistry.class) {
+            if (authKey.startsWith(AUTHPROVIDER_PROPERTY_PREFIX)) {
+                String className = System.getProperty(authKey);
+                try {
+                    Class<?> c = ZooKeeperServer.class.getClassLoader().loadClass(className);
+                    AuthenticationProvider ap = (AuthenticationProvider) c.getDeclaredConstructor().newInstance();
+                    authenticationProviders.put(ap.getScheme(), ap);
+                } catch (Exception e) {
+                    LOG.warn("Problems loading {}", className, e);
+                }
+            }
+        }
+    }
+
     public static ServerAuthenticationProvider getServerProvider(String scheme) {
         return WrappedAuthenticationProvider.wrap(getProvider(scheme));
     }
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/ClientSSLReloadTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/ClientSSLReloadTest.java
new file mode 100644
index 000000000..f9996c827
--- /dev/null
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/server/ClientSSLReloadTest.java
@@ -0,0 +1,229 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.
+ */
+package org.apache.zookeeper.server;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.security.Security;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import jline.internal.Log;
+import org.apache.commons.io.FileUtils;
+import org.apache.zookeeper.PortAssignment;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.ZKTestCase;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.client.ZKClientConfig;
+import org.apache.zookeeper.common.KeyStoreFileType;
+import org.apache.zookeeper.common.X509KeyType;
+import org.apache.zookeeper.common.X509TestContext;
+import org.apache.zookeeper.server.embedded.ExitHandler;
+import org.apache.zookeeper.server.embedded.ZooKeeperServerEmbedded;
+import org.apache.zookeeper.server.embedded.ZookeeperServeInfo;
+import org.apache.zookeeper.test.ClientBase;
+import org.bouncycastle.jce.provider.BouncyCastleProvider;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ClientSSLReloadTest extends ZKTestCase {
+    protected static final Logger LOG = LoggerFactory.getLogger(ClientSSLReloadTest.class);
+
+    private X509TestContext x509TestContext1;
+    private X509TestContext x509TestContext2;
+
+    private File dir1;
+    private File dir2;
+
+    private File keyStoreFile1;
+    private File trustStoreFile1;
+
+    private File keyStoreFile2;
+    private File trustStoreFile2;
+
+    @BeforeEach
+    public void setup() throws Exception {
+
+        dir1 = ClientBase.createEmptyTestDir();
+        dir2 = ClientBase.createEmptyTestDir();
+
+        Security.addProvider(new BouncyCastleProvider());
+
+        x509TestContext1 = X509TestContext.newBuilder()
+                .setTempDir(dir1)
+                .setKeyStoreKeyType(X509KeyType.EC)
+                .setTrustStoreKeyType(X509KeyType.EC)
+                .build();
+
+        x509TestContext2 = X509TestContext.newBuilder()
+                .setTempDir(dir2)
+                .setKeyStoreKeyType(X509KeyType.EC)
+                .setTrustStoreKeyType(X509KeyType.EC)
+                .build();
+
+        keyStoreFile1 = x509TestContext1.getKeyStoreFile(KeyStoreFileType.PEM);
+        trustStoreFile1 = x509TestContext1.getTrustStoreFile(KeyStoreFileType.PEM);
+
+        keyStoreFile2 = x509TestContext2.getKeyStoreFile(KeyStoreFileType.PEM);
+        trustStoreFile2 = x509TestContext2.getTrustStoreFile(KeyStoreFileType.PEM);
+
+        String testDataPath = System.getProperty("test.data.dir", "src/test/resources/data");
+    }
+
+    @AfterEach
+    public void teardown() throws Exception {
+        try {
+            FileUtils.deleteDirectory(dir1);
+            FileUtils.deleteDirectory(dir2);
+        } catch (IOException e) {
+            // ignore
+        }
+        Security.removeProvider(BouncyCastleProvider.PROVIDER_NAME);
+    }
+
+    /*
+     * This test performs certificate reload on a ZK server and checks the server presented certificate to the client.
+     * 1) setup() creates two sets of certificates to be used by the test.
+     * 2) Start the ZK server with TLS configuration ("client.certReload" config property will refresh the key and trust store for ClientX509Util at runtime)
+     * 3) ZK client will connect to the server on the secure client port using keyStoreFile1 and trustStoreFile1.
+     * 4) Update the keyStoreFile1 and trustStoreFile1 files in the filesystem with keyStoreFile2 and trustStoreFile2.
+     * 5) Till FileChangeWatcher thread is triggered & SSLContext options are reset, ZK client should continue to connect.
+     *    In Junit tests, FileChangeWatcher thread is not triggered immediately upon certifcate update in the filesystem.
+     * 6) Once the certficates are reloaded by the server, ZK client connect will fail.
+     * 7) Next, create a new ZK client with updated keystore & truststore paths (keyStoreFile2 and trustStoreFile2).
+     * 8) Server should accept the connection on the secure client port.
+     */
+    @Test
+    public void certficateReloadTest() throws Exception {
+
+        final Properties configZookeeper = getServerConfig();
+        try (ZooKeeperServerEmbedded zkServer = ZooKeeperServerEmbedded
+                .builder()
+                .baseDir(dir1.toPath())
+                .configuration(configZookeeper)
+                .exitHandler(ExitHandler.LOG_ONLY)
+                .build()) {
+            zkServer.start();
+            assertTrue(ClientBase.waitForServerUp(zkServer.getConnectionString(), 60000));
+            for (int i = 0; i < 100; i++) {
+                ZookeeperServeInfo.ServerInfo status = ZookeeperServeInfo.getStatus("StandaloneServer*");
+                if (status.isLeader() && status.isStandaloneMode()) {
+                    break;
+                }
+                Thread.sleep(100);
+            }
+            ZookeeperServeInfo.ServerInfo status = ZookeeperServeInfo.getStatus("StandaloneServer*");
+            assertTrue(status.isLeader());
+            assertTrue(status.isStandaloneMode());
+
+            CountDownLatch l = new CountDownLatch(1);
+            ZKClientConfig zKClientConfig = getZKClientConfig();
+            // ZK client object created which will connect with keyStoreFile1 and trustStoreFile1 to the server.
+            try (ZooKeeper zk = new ZooKeeper(zkServer.getSecureConnectionString(), 60000, (WatchedEvent event) -> {
+                switch (event.getState()) {
+                    case SyncConnected:
+                        l.countDown();
+                        break;
+                }
+            }, zKClientConfig)) {
+                assertTrue(zk.getClientConfig().getBoolean(ZKClientConfig.SECURE_CLIENT));
+                assertTrue(l.await(10, TimeUnit.SECONDS));
+            }
+
+            Log.info("Updating keyStore & trustStore files !!!!");
+            // Update the keyStoreFile1 and trustStoreFile1 files in the filesystem with keyStoreFile2 & trustStoreFile2
+            FileUtils.writeStringToFile(keyStoreFile1, FileUtils.readFileToString(keyStoreFile2, StandardCharsets.US_ASCII), StandardCharsets.US_ASCII, false);
+            FileUtils.writeStringToFile(trustStoreFile1, FileUtils.readFileToString(trustStoreFile2, StandardCharsets.US_ASCII), StandardCharsets.US_ASCII, false);
+
+            // Till FileChangeWatcher thread is triggered & SSLContext options are reset, ZK client should continue connecting.
+            for (int i = 0; i < 5; i++) {
+                CountDownLatch l2 = new CountDownLatch(1);
+                Thread.sleep(5000);
+                try (ZooKeeper zk = new ZooKeeper(zkServer.getSecureConnectionString(), 60000, (WatchedEvent event) -> {
+                    switch (event.getState()) {
+                        case SyncConnected:
+                            l.countDown();
+                            break;
+                    }
+                }, zKClientConfig)) {
+                    if (!l2.await(5, TimeUnit.SECONDS)) {
+                        LOG.error("Unable to connect to zk server");
+                        break;
+                    }
+                }
+            }
+            // Use the updated keyStore and trustStore paths when creating the client; Refreshed server should authenticate the client.
+            zKClientConfig.setProperty("zookeeper.ssl.keyStore.location", keyStoreFile2.getAbsolutePath());
+            zKClientConfig.setProperty("zookeeper.ssl.trustStore.location", trustStoreFile2.getAbsolutePath());
+            zKClientConfig.setProperty("zookeeper.ssl.keyStore.type", "PEM");
+            zKClientConfig.setProperty("zookeeper.ssl.trustStore.type", "PEM");
+            CountDownLatch l3 = new CountDownLatch(1);
+            try (ZooKeeper zk = new ZooKeeper(zkServer.getSecureConnectionString(), 60000, (WatchedEvent event) -> {
+                switch (event.getState()) {
+                    case SyncConnected:
+                        l3.countDown();
+                        break;
+                }
+            }, zKClientConfig)) {
+                assertTrue(zk.getClientConfig().getBoolean(ZKClientConfig.SECURE_CLIENT));
+                assertTrue(l3.await(10, TimeUnit.SECONDS));
+            }
+        }
+    }
+
+    private Properties getServerConfig() {
+        int clientPort = PortAssignment.unique();
+        int clientSecurePort = PortAssignment.unique();
+
+        final Properties configZookeeper = new Properties();
+        configZookeeper.put("clientPort", clientPort + "");
+        configZookeeper.put("secureClientPort", clientSecurePort + "");
+        configZookeeper.put("host", "localhost");
+        configZookeeper.put("ticktime", "4000");
+        configZookeeper.put("client.certReload", "true");
+        // TLS config fields
+        configZookeeper.put("ssl.keyStore.location", keyStoreFile1.getAbsolutePath());
+        configZookeeper.put("ssl.trustStore.location", trustStoreFile1.getAbsolutePath());
+        configZookeeper.put("ssl.keyStore.type", "PEM");
+        configZookeeper.put("ssl.trustStore.type", "PEM");
+        // Netty is required for TLS
+        configZookeeper.put("serverCnxnFactory", org.apache.zookeeper.server.NettyServerCnxnFactory.class.getName());
+        return configZookeeper;
+    }
+
+    private ZKClientConfig getZKClientConfig() throws IOException {
+        // Saving copies in JKS format to be used for client calls even after writeStringToFile overwrites keyStoreFile1 and trustStoreFile1
+        File clientKeyStore = x509TestContext1.getKeyStoreFile(KeyStoreFileType.JKS);
+        File clientTrustStore = x509TestContext1.getTrustStoreFile(KeyStoreFileType.JKS);
+
+        ZKClientConfig zKClientConfig = new ZKClientConfig();
+        zKClientConfig.setProperty("zookeeper.client.secure", "true");
+        zKClientConfig.setProperty("zookeeper.ssl.keyStore.location", clientKeyStore.getAbsolutePath());
+        zKClientConfig.setProperty("zookeeper.ssl.trustStore.location", clientTrustStore.getAbsolutePath());
+        zKClientConfig.setProperty("zookeeper.ssl.keyStore.type", "JKS");
+        zKClientConfig.setProperty("zookeeper.ssl.trustStore.type", "JKS");
+        // only netty supports TLS
+        zKClientConfig.setProperty("zookeeper.clientCnxnSocket", org.apache.zookeeper.ClientCnxnSocketNetty.class.getName());
+        return zKClientConfig;
+    }
+}