Skip to content

Commit

Permalink
ZOOKEEPER-4819 Fix can't seek for writable tls server if connected to…
Browse files Browse the repository at this point in the history
… 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.
  • Loading branch information
luoxiner committed Oct 14, 2024
1 parent 1880fc0 commit 33c19e3
Show file tree
Hide file tree
Showing 2 changed files with 141 additions and 32 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -19,13 +19,10 @@
package org.apache.zookeeper;

import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import java.io.BufferedReader;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.util.ArrayDeque;
Expand Down Expand Up @@ -67,10 +64,12 @@
import org.apache.zookeeper.ZooDefs.OpCode;
import org.apache.zookeeper.ZooKeeper.States;
import org.apache.zookeeper.ZooKeeper.WatchRegistration;
import org.apache.zookeeper.client.FourLetterWordMain;
import org.apache.zookeeper.client.HostProvider;
import org.apache.zookeeper.client.ZKClientConfig;
import org.apache.zookeeper.client.ZooKeeperSaslClient;
import org.apache.zookeeper.common.Time;
import org.apache.zookeeper.common.X509Exception;
import org.apache.zookeeper.proto.AuthPacket;
import org.apache.zookeeper.proto.ConnectRequest;
import org.apache.zookeeper.proto.Create2Response;
Expand Down Expand Up @@ -1301,44 +1300,19 @@ private void cleanAndNotifyState() {
private void pingRwServer() throws RWServerFoundException {
String result = null;
InetSocketAddress addr = hostProvider.next(0);
boolean useSecure = clientConfig.getBoolean(ZKClientConfig.SECURE_CLIENT);

LOG.info("Checking server {} for being r/w. Timeout {}", addr, pingRwTimeout);

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.getOutputStream().write("isro".getBytes());
sock.getOutputStream().flush();
sock.shutdownOutput();
br = new BufferedReader(new InputStreamReader(sock.getInputStream()));
result = br.readLine();
result = FourLetterWordMain.send4LetterWord(addr.getHostString(), addr.getPort(), "isro", useSecure, 1000);
} 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 {
if (sock != null) {
try {
sock.close();
} catch (IOException e) {
LOG.warn("Unexpected exception", e);
}
}
if (br != null) {
try {
br.close();
} catch (IOException e) {
LOG.warn("Unexpected exception", e);
}
}
}

if ("rw".equals(result)) {
if ("rw\n".equals(result)) {
pingRwTimeout = minPingRwTimeout;
// save the found address so that it's used during the next
// connection attempt
Expand Down
Original file line number Diff line number Diff line change
@@ -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.test;

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.CreateMode;
import org.apache.zookeeper.ZKTestCase;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
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.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");
}
}
}

0 comments on commit 33c19e3

Please sign in to comment.