Skip to content

Commit

Permalink
ZOOKEEPER-4819: Fix can't seek for writable tls server if connected t…
Browse files Browse the repository at this point in the history
…o readonly server

Reviewers: kezhuw
Author: luoxiner
Closes apache#2200 from luoxiner/ZOOKEEPER-4819

(cherry picked from commit 75f9781)
Signed-off-by: Kezhu Wang <[email protected]>
  • Loading branch information
luoxiner authored and kezhuw committed Nov 27, 2024
1 parent da1fc51 commit 47d23e5
Show file tree
Hide file tree
Showing 3 changed files with 203 additions and 40 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 @@ -1352,42 +1351,16 @@ private void pingRwServer() throws RWServerFoundException {
InetSocketAddress addr = hostProvider.next(0);

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", clientConfig, 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
Expand Up @@ -91,7 +91,56 @@ public static String send4LetterWord(
String cmd,
boolean secure,
int timeout) throws IOException, SSLContextException {
LOG.info("connecting to {} {}", host, port);
return send4LetterWord(host, port, cmd, secure, timeout, null);
}

/**
* Send the 4letterword
* @param host the destination host
* @param port the destination port
* @param cmd the 4letterword
* @param clientConfig client config
* @param timeout in milliseconds, maximum time to wait while connecting/reading data
* @return server response
* @throws SSLContextException
* @throws IOException
*/
public static String send4LetterWord(
String host,
int port,
String cmd,
ZKClientConfig clientConfig,
int timeout) throws SSLContextException, IOException {
boolean useSecure = clientConfig.getBoolean(ZKClientConfig.SECURE_CLIENT);
SSLContext sslContext = null;
if (useSecure) {
try (X509Util x509Util = new ClientX509Util()) {
sslContext = x509Util.createSSLContext(clientConfig);
}
}
return send4LetterWord(host, port, cmd, useSecure, timeout, sslContext);
}

/**
* Send the 4letterword
* @param host the destination host
* @param port the destination port
* @param cmd the 4letterword
* @param secure whether to use SSL
* @param timeout in milliseconds, maximum time to wait while connecting/reading data
* @param sslContext SSL context
* @return server response
* @throws java.io.IOException
* @throws SSLContextException
*/
public static String send4LetterWord(
String host,
int port,
String cmd,
boolean secure,
int timeout,
SSLContext sslContext) throws IOException, SSLContextException {
LOG.info("connecting to {}:{} (secure={})", host, port, secure);

Socket sock = null;
BufferedReader reader = null;
Expand All @@ -101,19 +150,23 @@ public static String send4LetterWord(
: new InetSocketAddress(InetAddress.getByName(null), port);
if (secure) {
LOG.info("using secure socket");
try (X509Util x509Util = new ClientX509Util()) {
SSLContext sslContext = x509Util.getDefaultSSLContext();
SSLSocketFactory socketFactory = sslContext.getSocketFactory();
SSLSocket sslSock = (SSLSocket) socketFactory.createSocket();
sslSock.connect(hostaddress, timeout);
sslSock.startHandshake();
sock = sslSock;
if (sslContext == null) {
try (X509Util x509Util = new ClientX509Util()) {
sslContext = x509Util.getDefaultSSLContext();
}
}
SSLSocketFactory socketFactory = sslContext.getSocketFactory();
SSLSocket sslSock = (SSLSocket) socketFactory.createSocket();
sslSock.connect(hostaddress, timeout);
sslSock.startHandshake();
sock = sslSock;
} else {
sock = new Socket();
sock.connect(hostaddress, timeout);
}
sock.setSoLinger(false, -1);
sock.setSoTimeout(timeout);
sock.setTcpNoDelay(true);
OutputStream outstream = sock.getOutputStream();
outstream.write(cmd.getBytes(UTF_8));
outstream.flush();
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,137 @@
/*
* 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;

public class ReadOnlyModeWithSSLTest extends ZKTestCase {

private static int CONNECTION_TIMEOUT = QuorumBase.CONNECTION_TIMEOUT;
private QuorumUtil qu = new QuorumUtil(1);
private ClientX509Util clientX509Util;
private ZKClientConfig clientConfig;

@BeforeEach
public void setUp() throws Exception {
clientX509Util = new ClientX509Util();
String testDataPath = System.getProperty("test.data.dir", "src/test/resources/data");
clientConfig = new ZKClientConfig();
clientConfig.setProperty(ZKClientConfig.SECURE_CLIENT, "true");
clientConfig.setProperty(ZKClientConfig.ZOOKEEPER_CLIENT_CNXN_SOCKET, "org.apache.zookeeper.ClientCnxnSocketNetty");
clientConfig.setProperty(clientX509Util.getSslKeystoreLocationProperty(), testDataPath + "/ssl/testKeyStore.jks");
clientConfig.setProperty(clientX509Util.getSslKeystorePasswdProperty(), "testpass");
clientConfig.setProperty(clientX509Util.getSslTruststoreLocationProperty(), testDataPath + "/ssl/testTrustStore.jks");
clientConfig.setProperty(clientX509Util.getSslTruststorePasswdProperty(), "testpass");
System.setProperty("readonlymode.enabled", "true");
System.setProperty(NettyServerCnxnFactory.PORT_UNIFICATION_KEY, Boolean.TRUE.toString());
System.setProperty(ServerCnxnFactory.ZOOKEEPER_SERVER_CNXN_FACTORY, "org.apache.zookeeper.server.NettyServerCnxnFactory");
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(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, clientConfig);
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 47d23e5

Please sign in to comment.