ZOOKEEPER-4819 Fix can't seek for writable tls server if connected to…

… readonly server

ClientCnxn::pingRwServer uses raw socket to issue "isro" 4lw command.
This results in unsuccessful handshake to tls server. Use SSLSocket
when zookeeper.client.secure is set to true.

zookeeper-server/src/main/java/org/apache/zookeeper/ClientCnxn.java

@@ -43,6 +43,9 @@
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.atomic.AtomicReference;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocket;
+import javax.net.ssl.SSLSocketFactory;
 import javax.security.auth.login.LoginException;
 import javax.security.sasl.SaslException;
 import org.apache.jute.BinaryInputArchive;
@@ -70,7 +73,10 @@
 import org.apache.zookeeper.client.HostProvider;
 import org.apache.zookeeper.client.ZKClientConfig;
 import org.apache.zookeeper.client.ZooKeeperSaslClient;
+import org.apache.zookeeper.common.ClientX509Util;
 import org.apache.zookeeper.common.Time;
+import org.apache.zookeeper.common.X509Exception;
+import org.apache.zookeeper.common.X509Util;
 import org.apache.zookeeper.proto.AuthPacket;
 import org.apache.zookeeper.proto.ConnectRequest;
 import org.apache.zookeeper.proto.Create2Response;
@@ -1307,18 +1313,15 @@ private void pingRwServer() throws RWServerFoundException {
             Socket sock = null;
             BufferedReader br = null;
             try {
-                sock = new Socket(addr.getHostString(), addr.getPort());
-                sock.setSoLinger(false, -1);
-                sock.setSoTimeout(1000);
-                sock.setTcpNoDelay(true);
+                sock = newSocket(addr);
                 sock.getOutputStream().write("isro".getBytes());
                 sock.getOutputStream().flush();
                 sock.shutdownOutput();
                 br = new BufferedReader(new InputStreamReader(sock.getInputStream()));
                 result = br.readLine();
             } catch (ConnectException e) {
                 // ignore, this just means server is not up
-            } catch (IOException e) {
+            } catch (IOException | X509Exception.SSLContextException e) {
                 // some unexpected error, warn about it
                 LOG.warn("Exception while seeking for r/w server.", e);
             } finally {
@@ -1348,6 +1351,28 @@ private void pingRwServer() throws RWServerFoundException {
             }
         }
 
+        private Socket newSocket(InetSocketAddress addr) throws IOException, X509Exception.SSLContextException {
+            boolean useSecure = clientConfig.getBoolean(ZKClientConfig.SECURE_CLIENT);
+            Socket sock;
+            if (useSecure) {
+                try (X509Util util = new ClientX509Util()) {
+                    SSLContext sslContext = util.createSSLContext(clientConfig);
+                    SSLSocketFactory socketFactory = sslContext.getSocketFactory();
+                    SSLSocket sslSock = (SSLSocket) socketFactory.createSocket();
+                    sslSock.connect(addr);
+                    sslSock.startHandshake();
+                    sock = sslSock;
+                }
+            } else {
+                sock = new Socket(addr.getHostString(), addr.getPort());
+            }
+
+            sock.setSoLinger(false, -1);
+            sock.setSoTimeout(1000);
+            sock.setTcpNoDelay(true);
+            return sock;
+        }
+
         private void cleanup() {
             clientCnxnSocket.cleanup();
             synchronized (pendingQueue) {

zookeeper-server/src/test/java/org/apache/zookeeper/test/ReadOnlyModeWithSSLTest.java

@@ -0,0 +1,135 @@
+/*
+ * 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.zookeeper;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import java.io.ByteArrayOutputStream;
+import java.io.LineNumberReader;
+import java.io.StringReader;
+import java.util.regex.Pattern;
+import org.apache.zookeeper.client.ZKClientConfig;
+import org.apache.zookeeper.common.ClientX509Util;
+import org.apache.zookeeper.common.StringUtils;
+import org.apache.zookeeper.server.NettyServerCnxnFactory;
+import org.apache.zookeeper.server.ServerCnxnFactory;
+import org.apache.zookeeper.test.ClientBase;
+import org.apache.zookeeper.test.LoggerTestTool;
+import org.apache.zookeeper.test.QuorumBase;
+import org.apache.zookeeper.test.QuorumUtil;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.slf4j.LoggerFactory;
+
+public class ReadOnlyModeWithSSLTest extends ZKTestCase{
+
+    private static final org.slf4j.Logger LOG = LoggerFactory.getLogger(ReadOnlyModeWithSSLTest.class);
+    private static int CONNECTION_TIMEOUT = QuorumBase.CONNECTION_TIMEOUT;
+    private QuorumUtil qu = new QuorumUtil(1);
+    private ClientX509Util clientX509Util;
+
+    @BeforeEach
+    public void setUp() throws Exception {
+        System.setProperty("readonlymode.enabled", "true");
+        System.setProperty(NettyServerCnxnFactory.PORT_UNIFICATION_KEY, Boolean.TRUE.toString());
+        clientX509Util = new ClientX509Util();
+        String testDataPath = System.getProperty("test.data.dir", "src/test/resources/data");
+        System.setProperty(ServerCnxnFactory.ZOOKEEPER_SERVER_CNXN_FACTORY, "org.apache.zookeeper.server.NettyServerCnxnFactory");
+        System.setProperty(ZKClientConfig.ZOOKEEPER_CLIENT_CNXN_SOCKET, "org.apache.zookeeper.ClientCnxnSocketNetty");
+        System.setProperty(ZKClientConfig.SECURE_CLIENT, "true");
+        System.setProperty(clientX509Util.getSslKeystoreLocationProperty(), testDataPath + "/ssl/testKeyStore.jks");
+        System.setProperty(clientX509Util.getSslKeystorePasswdProperty(), "testpass");
+        System.setProperty(clientX509Util.getSslTruststoreLocationProperty(), testDataPath + "/ssl/testTrustStore.jks");
+        System.setProperty(clientX509Util.getSslTruststorePasswdProperty(), "testpass");
+    }
+
+    @AfterEach
+    public void tearDown() throws Exception {
+        System.clearProperty(NettyServerCnxnFactory.PORT_UNIFICATION_KEY);
+        System.clearProperty(ServerCnxnFactory.ZOOKEEPER_SERVER_CNXN_FACTORY);
+        System.clearProperty(ZKClientConfig.ZOOKEEPER_CLIENT_CNXN_SOCKET);
+        System.clearProperty(ZKClientConfig.SECURE_CLIENT);
+        System.clearProperty(clientX509Util.getSslKeystoreLocationProperty());
+        System.clearProperty(clientX509Util.getSslKeystorePasswdProperty());
+        System.clearProperty(clientX509Util.getSslKeystorePasswdPathProperty());
+        System.clearProperty(clientX509Util.getSslTruststoreLocationProperty());
+        System.clearProperty(clientX509Util.getSslTruststorePasswdProperty());
+        System.clearProperty(clientX509Util.getSslTruststorePasswdPathProperty());
+        System.clearProperty(clientX509Util.getFipsModeProperty());
+        System.clearProperty(clientX509Util.getSslHostnameVerificationEnabledProperty());
+        System.clearProperty(clientX509Util.getSslProviderProperty());
+        clientX509Util.close();
+        System.setProperty("readonlymode.enabled", "false");
+        qu.tearDown();
+    }
+
+    /**
+     * Ensures that client seeks for r/w servers while it's connected to r/o
+     * server.
+     */
+    @Test
+    @Timeout(value = 90)
+    public void testSeekForRwServerWithSSL() throws Exception {
+        qu.enableLocalSession(true);
+        qu.startQuorum();
+
+        try (LoggerTestTool loggerTestTool = new LoggerTestTool("org.apache.zookeeper")) {
+            ByteArrayOutputStream os = loggerTestTool.getOutputStream();
+
+            qu.shutdown(2);
+            ClientBase.CountdownWatcher watcher = new ClientBase.CountdownWatcher();
+            ZooKeeper zk = new ZooKeeper(qu.getConnString(), CONNECTION_TIMEOUT, watcher, true);
+            watcher.waitForConnected(CONNECTION_TIMEOUT);
+
+            // if we don't suspend a peer it will rejoin a quorum
+            qu.getPeer(1).peer
+                    .setSuspended(true);
+
+            // start two servers to form a quorum; client should detect this and
+            // connect to one of them
+            watcher.reset();
+            qu.start(2);
+            qu.start(3);
+            ClientBase.waitForServerUp(qu.getConnString(), 2000);
+            watcher.waitForConnected(CONNECTION_TIMEOUT);
+            zk.create("/test", "test".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+
+            // resume poor fellow
+            qu.getPeer(1).peer
+                    .setSuspended(false);
+
+            String log = os.toString();
+            assertFalse(StringUtils.isEmpty(log), "OutputStream doesn't have any log messages");
+
+            LineNumberReader r = new LineNumberReader(new StringReader(log));
+            String line;
+            Pattern p = Pattern.compile(".*Majority server found.*");
+            boolean found = false;
+            while ((line = r.readLine()) != null) {
+                if (p.matcher(line).matches()) {
+                    found = true;
+                    break;
+                }
+            }
+            assertTrue(found, "Majority server wasn't found while connected to r/o server");
+        }
+    }
+}