Refactored and Added Test Case

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

@@ -18,8 +18,6 @@
  */
 package org.apache.pulsar.broker.service;
 
-import java.util.concurrent.TimeUnit;
-
 import org.apache.pulsar.broker.PulsarService;
 import org.apache.pulsar.broker.ServiceConfiguration;
 import org.apache.pulsar.common.api.ByteBufPair;
@@ -51,8 +49,8 @@ public PulsarChannelInitializer(PulsarService pulsar, boolean enableTLS) throws
             sslCtxRefresher = new ServerSslContextRefresher(serviceConfig.isTlsAllowInsecureConnection(),
                     serviceConfig.getTlsTrustCertsFilePath(), serviceConfig.getTlsCertificateFilePath(),
                     serviceConfig.getTlsKeyFilePath(), serviceConfig.getTlsCiphers(), serviceConfig.getTlsProtocols(),
-                    serviceConfig.isTlsRequireTrustedClientCertOnConnect(), pulsar.getExecutor(),
-                    serviceConfig.getCertRefreshCheckDurationInMins(), TimeUnit.MINUTES);
+                    serviceConfig.isTlsRequireTrustedClientCertOnConnect(),
+                    serviceConfig.getCertRefreshCheckDurationInMins());
         } else {
             this.sslCtxRefresher = null;
         }

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/AuthenticationDataProvider.java

@@ -40,22 +40,6 @@ public interface AuthenticationDataProvider extends Serializable {
     default boolean hasDataForTls() {
         return false;
     }
-
-    /**
-     * 
-     * @return a Client Certificate File Path, or null if data is not available
-     */
-    default String getCertFilePath() {
-        return null;
-    }
-
-    /**
-     * 
-     * @return a Client Key Certificate File Path, or null if data is not available
-     */
-    default String getKeyFilePath() {
-        return null;
-    }
 
     /**
      * 

pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationDataTls.java

@@ -20,30 +20,29 @@
 
 import java.security.KeyManagementException;
 import java.security.PrivateKey;
+import java.security.cert.Certificate;
 import java.security.cert.X509Certificate;
 
 import org.apache.pulsar.client.api.AuthenticationDataProvider;
+import org.apache.pulsar.common.util.FileModifiedTimeUpdater;
 import org.apache.pulsar.common.util.SecurityUtility;
-
-import lombok.Getter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class AuthenticationDataTls implements AuthenticationDataProvider {
+    protected X509Certificate[] tlsCertificates;
+    protected PrivateKey tlsPrivateKey;
+    protected FileModifiedTimeUpdater certFile, keyFile;
 
-    @Getter
-    protected final X509Certificate[] tlsCertificates;
-    @Getter
-    protected final PrivateKey tlsPrivateKey;
-    @Getter
-    private String certFilePath, keyFilePath;
     public AuthenticationDataTls(String certFilePath, String keyFilePath) throws KeyManagementException {
         if (certFilePath == null) {
             throw new IllegalArgumentException("certFilePath must not be null");
         }
         if (keyFilePath == null) {
             throw new IllegalArgumentException("keyFilePath must not be null");
         }
-        this.certFilePath = certFilePath;
-        this.keyFilePath = keyFilePath;
+        this.certFile = new FileModifiedTimeUpdater(certFilePath);
+        this.keyFile = new FileModifiedTimeUpdater(keyFilePath);
         this.tlsCertificates = SecurityUtility.loadCertificatesFromPemFile(certFilePath);
         this.tlsPrivateKey = SecurityUtility.loadPrivateKeyFromPemFile(keyFilePath);
     }
@@ -57,4 +56,29 @@ public boolean hasDataForTls() {
         return true;
     }
 
+    @Override
+    public Certificate[] getTlsCertificates() {
+        if (this.certFile.checkAndRefresh()) {
+            try {
+                this.tlsCertificates = SecurityUtility.loadCertificatesFromPemFile(certFile.getFileName());
+            } catch (KeyManagementException e) {
+                LOG.error("Unable to refresh authData for cert {}: ", certFile.getFileName(), e);
+            }
+        }
+        return this.tlsCertificates;
+    }
+
+    @Override
+    public PrivateKey getTlsPrivateKey() {
+        if (this.keyFile.checkAndRefresh()) {
+            try {
+                this.tlsPrivateKey = SecurityUtility.loadPrivateKeyFromPemFile(keyFile.getFileName());
+            } catch (KeyManagementException e) {
+                LOG.error("Unable to refresh authData for cert {}: ", keyFile.getFileName(), e);
+            }
+        }
+        return this.tlsPrivateKey;
+    }
+
+    private static final Logger LOG = LoggerFactory.getLogger(AuthenticationDataTls.class);
 }

pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationTls.java

@@ -49,6 +49,10 @@ public class AuthenticationTls implements Authentication, EncodedAuthenticationP
 
     public AuthenticationTls() {
     }
+
+    public AuthenticationTls(String certFilePath, String keyFilePath, long refresh) {
+
+    }
 
     public AuthenticationTls(String certFilePath, String keyFilePath) {
         this.certFilePath = certFilePath;

pulsar-common/src/main/java/org/apache/pulsar/common/util/ClientSslContextRefresher.java

@@ -18,71 +18,51 @@
  */
 package org.apache.pulsar.common.util;
 
-import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.security.GeneralSecurityException;
-import java.security.KeyManagementException;
 import java.security.cert.X509Certificate;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-
-import javax.net.ssl.SSLException;
 
 import org.apache.pulsar.client.api.AuthenticationDataProvider;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import io.netty.handler.ssl.SslContext;
 
-public class ClientSslContextRefresher extends SslContextRefresher {
+public class ClientSslContextRefresher {
     private volatile SslContext sslContext;
     private boolean tlsAllowInsecureConnection;
     private String tlsTrustCertsFilePath;
     private AuthenticationDataProvider authData;
 
     public ClientSslContextRefresher(boolean allowInsecure, String trustCertsFilePath,
-            AuthenticationDataProvider authData, ScheduledExecutorService eventLoopGroup, long delay, TimeUnit timeUnit)
-            throws IOException, GeneralSecurityException {
-        super(eventLoopGroup, delay, timeUnit);
+            AuthenticationDataProvider authData) throws IOException, GeneralSecurityException {
         this.tlsAllowInsecureConnection = allowInsecure;
         this.tlsTrustCertsFilePath = trustCertsFilePath;
         this.authData = authData;
 
         if (authData != null && authData.hasDataForTls()) {
-            registerFile(authData.getCertFilePath(), authData.getKeyFilePath());
             this.sslContext = SecurityUtility.createNettySslContextForClient(this.tlsAllowInsecureConnection,
                     this.tlsTrustCertsFilePath, (X509Certificate[]) authData.getTlsCertificates(),
                     authData.getTlsPrivateKey());
-            if (delay > 0) {
-                run();
-            }
         } else {
             this.sslContext = SecurityUtility.createNettySslContextForClient(this.tlsAllowInsecureConnection,
                     this.tlsTrustCertsFilePath);
         }
     }
 
-    private static final Logger LOG = LoggerFactory.getLogger(ClientSslContextRefresher.class);
+    public SslContext get() {
+        if (authData != null && authData.hasDataForTls()) {
+            try {
+                this.sslContext = SecurityUtility.createNettySslContextForClient(this.tlsAllowInsecureConnection,
+                        this.tlsTrustCertsFilePath, (X509Certificate[]) authData.getTlsCertificates(),
+                        authData.getTlsPrivateKey());
+            } catch (GeneralSecurityException | IOException e) {
+                LOG.error("Exception occured while trying to refresh sslContext: ", e);
+            }
 
-    @Override
-    public void buildSSLContext() {
-        try {
-            buildSSLContextWithException();
-        } catch (GeneralSecurityException | IOException e) {
-            LOG.error("Error occured while trying to create sslContext - using previous one.");
-            return;
         }
-    }
-
-    public void buildSSLContextWithException() throws KeyManagementException, SSLException, FileNotFoundException, GeneralSecurityException, IOException {
-        this.sslContext = SecurityUtility.createNettySslContextForClient(this.tlsAllowInsecureConnection,
-                this.tlsTrustCertsFilePath, 
-                (X509Certificate[]) SecurityUtility.loadCertificatesFromPemFile(authData.getCertFilePath()),
-                SecurityUtility.loadPrivateKeyFromPemFile(authData.getKeyFilePath()));
-    }
-
-    @Override
-    public SslContext get() {
         return sslContext;
     }
+
+    private static final Logger LOG = LoggerFactory.getLogger(ClientSslContextRefresher.class);
 }

pulsar-common/src/main/java/org/apache/pulsar/common/util/FileModifiedTimeUpdater.java

@@ -0,0 +1,66 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.util;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.attribute.FileTime;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import lombok.Getter;
+
+public class FileModifiedTimeUpdater {
+    @Getter
+    String fileName;
+    @Getter
+    FileTime lastModifiedTime;
+
+    public FileModifiedTimeUpdater(String fileName) {
+        this.fileName = fileName;
+        this.lastModifiedTime = updateLastModifiedTime();
+    }
+
+    private FileTime updateLastModifiedTime() {
+        if (fileName != null) {
+            Path p = Paths.get(fileName);
+            try {
+                return Files.getLastModifiedTime(p);
+            } catch (IOException e) {
+                LOG.error("Unable to fetch lastModified time for file {}: ", fileName, e);
+            }
+        }
+        return null;
+    }
+
+    public boolean checkAndRefresh() {
+        boolean ret = false;
+        FileTime newLastModifiedTime = updateLastModifiedTime();
+        if (newLastModifiedTime != null && !newLastModifiedTime.equals(lastModifiedTime)) {
+            this.lastModifiedTime = newLastModifiedTime;
+            ret = true;
+        }
+        return ret;
+    }
+
+    private static final Logger LOG = LoggerFactory.getLogger(FileModifiedTimeUpdater.class);
+}

pulsar-common/src/main/java/org/apache/pulsar/common/util/ServerSslContextRefresher.java

@@ -22,7 +22,6 @@
 import java.io.IOException;
 import java.security.GeneralSecurityException;
 import java.util.Set;
-import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
 import javax.net.ssl.SSLException;
@@ -32,60 +31,55 @@
 
 import io.netty.handler.ssl.SslContext;
 
-public class ServerSslContextRefresher extends SslContextRefresher {
+public class ServerSslContextRefresher {
     private final boolean tlsAllowInsecureConnection;
-    private final String tlsTrustCertsFilePath;
-    private final String tlsCertificateFilePath;
-    private final String tlsKeyFilePath;
+    private final FileModifiedTimeUpdater tlsTrustCertsFilePath, tlsCertificateFilePath, tlsKeyFilePath;
     private final Set<String> tlsCiphers;
     private final Set<String> tlsProtocols;
     private final boolean tlsRequireTrustedClientCertOnConnect;
+    private final long delayInMins;
+    private long nextRefreshTimeInMins;
     private volatile SslContext sslContext;
 
     public ServerSslContextRefresher(boolean allowInsecure, String trustCertsFilePath, String certificateFilePath,
             String keyFilePath, Set<String> ciphers, Set<String> protocols, boolean requireTrustedClientCertOnConnect,
-            ScheduledExecutorService eventLoopGroup, long delay, TimeUnit timeUnit)
-            throws SSLException, FileNotFoundException, GeneralSecurityException, IOException {
-        super(eventLoopGroup, delay, timeUnit);
+            long delayInMins) throws SSLException, FileNotFoundException, GeneralSecurityException, IOException {
         this.tlsAllowInsecureConnection = allowInsecure;
-        this.tlsTrustCertsFilePath = trustCertsFilePath;
-        this.tlsCertificateFilePath = certificateFilePath;
-        this.tlsKeyFilePath = keyFilePath;
+        this.tlsTrustCertsFilePath = new FileModifiedTimeUpdater(trustCertsFilePath);
+        this.tlsCertificateFilePath = new FileModifiedTimeUpdater(certificateFilePath);
+        this.tlsKeyFilePath = new FileModifiedTimeUpdater(keyFilePath);
         this.tlsCiphers = ciphers;
         this.tlsProtocols = protocols;
         this.tlsRequireTrustedClientCertOnConnect = requireTrustedClientCertOnConnect;
+        this.delayInMins = delayInMins;
+        this.nextRefreshTimeInMins = TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis()) + delayInMins;
 
-        registerFile(tlsTrustCertsFilePath, tlsCertificateFilePath, tlsKeyFilePath);
-
-        buildSSLContextWithException();
+        buildSSLContext();
 
         if (LOG.isDebugEnabled()) {
-            LOG.debug("Certs will be refreshed every {} minutes", delay);
-        }
-
-        if (delay > 0) {
-            run();
+            LOG.debug("Certs will be refreshed every {} minutes", delayInMins);
         }
     }
 
-    @Override
-    public void buildSSLContext() {
-        try {
-            buildSSLContextWithException();
-        } catch (GeneralSecurityException | IOException e) {
-            LOG.error("Error occured while trying to create sslContext - using previous one.");
-            return;
-        }
-    }
-
-    public void buildSSLContextWithException() throws SSLException, FileNotFoundException, GeneralSecurityException, IOException {
+    public void buildSSLContext() throws SSLException, FileNotFoundException, GeneralSecurityException, IOException {
         this.sslContext = SecurityUtility.createNettySslContextForServer(tlsAllowInsecureConnection,
-                tlsTrustCertsFilePath, tlsCertificateFilePath, tlsKeyFilePath, tlsCiphers, tlsProtocols,
-                tlsRequireTrustedClientCertOnConnect);
+                tlsTrustCertsFilePath.getFileName(), tlsCertificateFilePath.getFileName(), tlsKeyFilePath.getFileName(),
+                tlsCiphers, tlsProtocols, tlsRequireTrustedClientCertOnConnect);
     }
 
-    @Override
     public SslContext get() {
+        long nowInSeconds = TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis());
+        if (nextRefreshTimeInMins > nowInSeconds) {
+            nextRefreshTimeInMins = nowInSeconds + delayInMins;
+            if (tlsTrustCertsFilePath.checkAndRefresh() || tlsCertificateFilePath.checkAndRefresh()
+                    || tlsKeyFilePath.checkAndRefresh()) {
+                try {
+                    buildSSLContext();
+                } catch (GeneralSecurityException | IOException e) {
+                    LOG.error("Execption while trying to refresh ssl Context: ", e);
+                }
+            }
+        }
         return this.sslContext;
     }