You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by su...@apache.org on 2016/09/14 02:43:44 UTC
[06/50] [abbrv] hadoop git commit: HADOOP-10940. RPC client does no
bounds checking of responses. Contributed by Daryn Sharp.
HADOOP-10940. RPC client does no bounds checking of responses. Contributed by Daryn Sharp.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/d4d07687
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/d4d07687
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/d4d07687
Branch: refs/heads/YARN-2915
Commit: d4d076876a8d0002bd3a73491d8459d11cb4896c
Parents: baab489
Author: Kihwal Lee <ki...@apache.org>
Authored: Fri Sep 9 10:39:35 2016 -0500
Committer: Kihwal Lee <ki...@apache.org>
Committed: Fri Sep 9 10:39:35 2016 -0500
----------------------------------------------------------------------
.../dev-support/findbugsExcludeFile.xml | 10 +-
.../hadoop/fs/CommonConfigurationKeys.java | 12 +-
.../main/java/org/apache/hadoop/ipc/Client.java | 178 +++++++++++++------
.../apache/hadoop/security/SaslRpcClient.java | 25 +--
.../src/main/resources/core-default.xml | 17 +-
.../java/org/apache/hadoop/ipc/TestIPC.java | 86 ++++++++-
6 files changed, 246 insertions(+), 82 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/d4d07687/hadoop-common-project/hadoop-common/dev-support/findbugsExcludeFile.xml
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/dev-support/findbugsExcludeFile.xml b/hadoop-common-project/hadoop-common/dev-support/findbugsExcludeFile.xml
index b650eae..ec7c396 100644
--- a/hadoop-common-project/hadoop-common/dev-support/findbugsExcludeFile.xml
+++ b/hadoop-common-project/hadoop-common/dev-support/findbugsExcludeFile.xml
@@ -44,7 +44,7 @@
-->
<Match>
<Class name="org.apache.hadoop.ipc.Client$Connection" />
- <Field name="out" />
+ <Field name="ipcStreams" />
<Bug pattern="IS2_INCONSISTENT_SYNC" />
</Match>
<!--
@@ -341,13 +341,7 @@
<Method name="removeRenewAction" />
<Bug pattern="BC_UNCONFIRMED_CAST" />
</Match>
-
- <!-- Inconsistent synchronization flagged by findbugs is not valid. -->
- <Match>
- <Class name="org.apache.hadoop.ipc.Client$Connection" />
- <Field name="in" />
- <Bug pattern="IS2_INCONSISTENT_SYNC" />
- </Match>
+
<!--
The switch condition for INITIATE is expected to fallthru to RESPONSE
to process initial sasl response token included in the INITIATE
http://git-wip-us.apache.org/repos/asf/hadoop/blob/d4d07687/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonConfigurationKeys.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonConfigurationKeys.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonConfigurationKeys.java
index 76cf9cc..2b530f0 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonConfigurationKeys.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonConfigurationKeys.java
@@ -78,12 +78,20 @@ public class CommonConfigurationKeys extends CommonConfigurationKeysPublic {
/** Default value for IPC_SERVER_RPC_READ_CONNECTION_QUEUE_SIZE */
public static final int IPC_SERVER_RPC_READ_CONNECTION_QUEUE_SIZE_DEFAULT =
100;
-
+
+ /** Max request size a server will accept. */
public static final String IPC_MAXIMUM_DATA_LENGTH =
"ipc.maximum.data.length";
-
+ /** Default value for IPC_MAXIMUM_DATA_LENGTH. */
public static final int IPC_MAXIMUM_DATA_LENGTH_DEFAULT = 64 * 1024 * 1024;
+ /** Max response size a client will accept. */
+ public static final String IPC_MAXIMUM_RESPONSE_LENGTH =
+ "ipc.maximum.response.length";
+ /** Default value for IPC_MAXIMUM_RESPONSE_LENGTH. */
+ public static final int IPC_MAXIMUM_RESPONSE_LENGTH_DEFAULT =
+ 128 * 1024 * 1024;
+
/** How many calls per handler are allowed in the queue. */
public static final String IPC_SERVER_HANDLER_QUEUE_SIZE_KEY =
"ipc.server.handler.queue.size";
http://git-wip-us.apache.org/repos/asf/hadoop/blob/d4d07687/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java
index 567b932..94f1945 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java
@@ -32,6 +32,7 @@ import org.apache.hadoop.fs.CommonConfigurationKeys;
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableUtils;
import org.apache.hadoop.io.retry.RetryPolicies;
import org.apache.hadoop.io.retry.RetryPolicy;
import org.apache.hadoop.io.retry.RetryPolicy.RetryAction;
@@ -413,8 +414,8 @@ public class Client implements AutoCloseable {
private SaslRpcClient saslRpcClient;
private Socket socket = null; // connected socket
- private DataInputStream in;
- private DataOutputStream out;
+ private IpcStreams ipcStreams;
+ private final int maxResponseLength;
private final int rpcTimeout;
private int maxIdleTime; //connections will be culled if it was idle for
//maxIdleTime msecs
@@ -426,8 +427,8 @@ public class Client implements AutoCloseable {
private final boolean doPing; //do we need to send ping message
private final int pingInterval; // how often sends ping to the server
private final int soTimeout; // used by ipc ping and rpc timeout
- private ResponseBuffer pingRequest; // ping message
-
+ private byte[] pingRequest; // ping message
+
// currently active calls
private Hashtable<Integer, Call> calls = new Hashtable<Integer, Call>();
private AtomicLong lastActivity = new AtomicLong();// last I/O activity time
@@ -446,6 +447,9 @@ public class Client implements AutoCloseable {
0,
new UnknownHostException());
}
+ this.maxResponseLength = remoteId.conf.getInt(
+ CommonConfigurationKeys.IPC_MAXIMUM_RESPONSE_LENGTH,
+ CommonConfigurationKeys.IPC_MAXIMUM_RESPONSE_LENGTH_DEFAULT);
this.rpcTimeout = remoteId.getRpcTimeout();
this.maxIdleTime = remoteId.getMaxIdleTime();
this.connectionRetryPolicy = remoteId.connectionRetryPolicy;
@@ -456,12 +460,13 @@ public class Client implements AutoCloseable {
this.doPing = remoteId.getDoPing();
if (doPing) {
// construct a RPC header with the callId as the ping callId
- pingRequest = new ResponseBuffer();
+ ResponseBuffer buf = new ResponseBuffer();
RpcRequestHeaderProto pingHeader = ProtoUtil
.makeRpcRequestHeader(RpcKind.RPC_PROTOCOL_BUFFER,
OperationProto.RPC_FINAL_PACKET, PING_CALL_ID,
RpcConstants.INVALID_RETRY_COUNT, clientId);
- pingHeader.writeDelimitedTo(pingRequest);
+ pingHeader.writeDelimitedTo(buf);
+ pingRequest = buf.toByteArray();
}
this.pingInterval = remoteId.getPingInterval();
if (rpcTimeout > 0) {
@@ -596,15 +601,15 @@ public class Client implements AutoCloseable {
}
return false;
}
-
- private synchronized AuthMethod setupSaslConnection(final InputStream in2,
- final OutputStream out2) throws IOException {
+
+ private synchronized AuthMethod setupSaslConnection(IpcStreams streams)
+ throws IOException {
// Do not use Client.conf here! We must use ConnectionId.conf, since the
// Client object is cached and shared between all RPC clients, even those
// for separate services.
saslRpcClient = new SaslRpcClient(remoteId.getTicket(),
remoteId.getProtocol(), remoteId.getAddress(), remoteId.conf);
- return saslRpcClient.saslConnect(in2, out2);
+ return saslRpcClient.saslConnect(streams);
}
/**
@@ -770,12 +775,9 @@ public class Client implements AutoCloseable {
Random rand = null;
while (true) {
setupConnection();
- InputStream inStream = NetUtils.getInputStream(socket);
- OutputStream outStream = NetUtils.getOutputStream(socket);
- writeConnectionHeader(outStream);
+ ipcStreams = new IpcStreams(socket, maxResponseLength);
+ writeConnectionHeader(ipcStreams);
if (authProtocol == AuthProtocol.SASL) {
- final InputStream in2 = inStream;
- final OutputStream out2 = outStream;
UserGroupInformation ticket = remoteId.getTicket();
if (ticket.getRealUser() != null) {
ticket = ticket.getRealUser();
@@ -786,7 +788,7 @@ public class Client implements AutoCloseable {
@Override
public AuthMethod run()
throws IOException, InterruptedException {
- return setupSaslConnection(in2, out2);
+ return setupSaslConnection(ipcStreams);
}
});
} catch (IOException ex) {
@@ -805,8 +807,7 @@ public class Client implements AutoCloseable {
}
if (authMethod != AuthMethod.SIMPLE) {
// Sasl connect is successful. Let's set up Sasl i/o streams.
- inStream = saslRpcClient.getInputStream(inStream);
- outStream = saslRpcClient.getOutputStream(outStream);
+ ipcStreams.setSaslClient(saslRpcClient);
// for testing
remoteId.saslQop =
(String)saslRpcClient.getNegotiatedProperty(Sasl.QOP);
@@ -825,18 +826,11 @@ public class Client implements AutoCloseable {
}
}
}
-
+
if (doPing) {
- inStream = new PingInputStream(inStream);
+ ipcStreams.setInputStream(new PingInputStream(ipcStreams.in));
}
- this.in = new DataInputStream(new BufferedInputStream(inStream));
- // SASL may have already buffered the stream
- if (!(outStream instanceof BufferedOutputStream)) {
- outStream = new BufferedOutputStream(outStream);
- }
- this.out = new DataOutputStream(outStream);
-
writeConnectionContext(remoteId, authMethod);
// update last activity time
@@ -950,17 +944,28 @@ public class Client implements AutoCloseable {
* | AuthProtocol (1 byte) |
* +----------------------------------+
*/
- private void writeConnectionHeader(OutputStream outStream)
+ private void writeConnectionHeader(IpcStreams streams)
throws IOException {
- DataOutputStream out = new DataOutputStream(new BufferedOutputStream(outStream));
- // Write out the header, version and authentication method
- out.write(RpcConstants.HEADER.array());
- out.write(RpcConstants.CURRENT_VERSION);
- out.write(serviceClass);
- out.write(authProtocol.callId);
- out.flush();
+ // Write out the header, version and authentication method.
+ // The output stream is buffered but we must not flush it yet. The
+ // connection setup protocol requires the client to send multiple
+ // messages before reading a response.
+ //
+ // insecure: send header+context+call, read
+ // secure : send header+negotiate, read, (sasl), context+call, read
+ //
+ // The client must flush only when it's prepared to read. Otherwise
+ // "broken pipe" exceptions occur if the server closes the connection
+ // before all messages are sent.
+ final DataOutputStream out = streams.out;
+ synchronized (out) {
+ out.write(RpcConstants.HEADER.array());
+ out.write(RpcConstants.CURRENT_VERSION);
+ out.write(serviceClass);
+ out.write(authProtocol.callId);
+ }
}
-
+
/* Write the connection context header for each connection
* Out is not synchronized because only the first thread does this.
*/
@@ -976,12 +981,17 @@ public class Client implements AutoCloseable {
.makeRpcRequestHeader(RpcKind.RPC_PROTOCOL_BUFFER,
OperationProto.RPC_FINAL_PACKET, CONNECTION_CONTEXT_CALL_ID,
RpcConstants.INVALID_RETRY_COUNT, clientId);
+ // do not flush. the context and first ipc call request must be sent
+ // together to avoid possibility of broken pipes upon authz failure.
+ // see writeConnectionHeader
final ResponseBuffer buf = new ResponseBuffer();
connectionContextHeader.writeDelimitedTo(buf);
message.writeDelimitedTo(buf);
- buf.writeTo(out);
+ synchronized (ipcStreams.out) {
+ ipcStreams.sendRequest(buf.toByteArray());
+ }
}
-
+
/* wait till someone signals us to start reading RPC response or
* it is idle too long, it is marked as to be closed,
* or the client is marked as not running.
@@ -1024,9 +1034,9 @@ public class Client implements AutoCloseable {
long curTime = Time.now();
if ( curTime - lastActivity.get() >= pingInterval) {
lastActivity.set(curTime);
- synchronized (out) {
- pingRequest.writeTo(out);
- out.flush();
+ synchronized (ipcStreams.out) {
+ ipcStreams.sendRequest(pingRequest);
+ ipcStreams.flush();
}
}
}
@@ -1092,15 +1102,16 @@ public class Client implements AutoCloseable {
@Override
public void run() {
try {
- synchronized (Connection.this.out) {
+ synchronized (ipcStreams.out) {
if (shouldCloseConnection.get()) {
return;
}
if (LOG.isDebugEnabled()) {
LOG.debug(getName() + " sending #" + call.id);
}
- buf.writeTo(out); // RpcRequestHeader + RpcRequest
- out.flush();
+ // RpcRequestHeader + RpcRequest
+ ipcStreams.sendRequest(buf.toByteArray());
+ ipcStreams.flush();
}
} catch (IOException e) {
// exception at this point would leave the connection in an
@@ -1141,10 +1152,7 @@ public class Client implements AutoCloseable {
touch();
try {
- int totalLen = in.readInt();
- ByteBuffer bb = ByteBuffer.allocate(totalLen);
- in.readFully(bb.array());
-
+ ByteBuffer bb = ipcStreams.readResponse();
RpcWritable.Buffer packet = RpcWritable.Buffer.wrap(bb);
RpcResponseHeaderProto header =
packet.getValue(RpcResponseHeaderProto.getDefaultInstance());
@@ -1209,8 +1217,7 @@ public class Client implements AutoCloseable {
connections.remove(remoteId, this);
// close the streams and therefore the socket
- IOUtils.closeStream(out);
- IOUtils.closeStream(in);
+ IOUtils.closeStream(ipcStreams);
disposeSasl();
// clean up all calls
@@ -1739,4 +1746,75 @@ public class Client implements AutoCloseable {
public void close() throws Exception {
stop();
}
+
+ /** Manages the input and output streams for an IPC connection.
+ * Only exposed for use by SaslRpcClient.
+ */
+ @InterfaceAudience.Private
+ public static class IpcStreams implements Closeable, Flushable {
+ private DataInputStream in;
+ public DataOutputStream out;
+ private int maxResponseLength;
+ private boolean firstResponse = true;
+
+ IpcStreams(Socket socket, int maxResponseLength) throws IOException {
+ this.maxResponseLength = maxResponseLength;
+ setInputStream(
+ new BufferedInputStream(NetUtils.getInputStream(socket)));
+ setOutputStream(
+ new BufferedOutputStream(NetUtils.getOutputStream(socket)));
+ }
+
+ void setSaslClient(SaslRpcClient client) throws IOException {
+ setInputStream(client.getInputStream(in));
+ setOutputStream(client.getOutputStream(out));
+ }
+
+ private void setInputStream(InputStream is) {
+ this.in = (is instanceof DataInputStream)
+ ? (DataInputStream)is : new DataInputStream(is);
+ }
+
+ private void setOutputStream(OutputStream os) {
+ this.out = (os instanceof DataOutputStream)
+ ? (DataOutputStream)os : new DataOutputStream(os);
+ }
+
+ public ByteBuffer readResponse() throws IOException {
+ int length = in.readInt();
+ if (firstResponse) {
+ firstResponse = false;
+ // pre-rpcv9 exception, almost certainly a version mismatch.
+ if (length == -1) {
+ in.readInt(); // ignore fatal/error status, it's fatal for us.
+ throw new RemoteException(WritableUtils.readString(in),
+ WritableUtils.readString(in));
+ }
+ }
+ if (length <= 0) {
+ throw new RpcException("RPC response has invalid length");
+ }
+ if (maxResponseLength > 0 && length > maxResponseLength) {
+ throw new RpcException("RPC response exceeds maximum data length");
+ }
+ ByteBuffer bb = ByteBuffer.allocate(length);
+ in.readFully(bb.array());
+ return bb;
+ }
+
+ public void sendRequest(byte[] buf) throws IOException {
+ out.write(buf);
+ }
+
+ @Override
+ public void flush() throws IOException {
+ out.flush();
+ }
+
+ @Override
+ public void close() {
+ IOUtils.closeStream(out);
+ IOUtils.closeStream(in);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/d4d07687/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/SaslRpcClient.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/SaslRpcClient.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/SaslRpcClient.java
index 60ae3b0..cd942b7 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/SaslRpcClient.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/SaslRpcClient.java
@@ -18,11 +18,9 @@
package org.apache.hadoop.security;
-import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
-import java.io.DataOutputStream;
import java.io.FilterInputStream;
import java.io.FilterOutputStream;
import java.io.IOException;
@@ -53,6 +51,7 @@ import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.GlobPattern;
+import org.apache.hadoop.ipc.Client.IpcStreams;
import org.apache.hadoop.ipc.RPC.RpcKind;
import org.apache.hadoop.ipc.RemoteException;
import org.apache.hadoop.ipc.ResponseBuffer;
@@ -353,24 +352,16 @@ public class SaslRpcClient {
* @return AuthMethod used to negotiate the connection
* @throws IOException
*/
- public AuthMethod saslConnect(InputStream inS, OutputStream outS)
- throws IOException {
- DataInputStream inStream = new DataInputStream(new BufferedInputStream(inS));
- DataOutputStream outStream = new DataOutputStream(new BufferedOutputStream(
- outS));
-
+ public AuthMethod saslConnect(IpcStreams ipcStreams) throws IOException {
// redefined if/when a SASL negotiation starts, can be queried if the
// negotiation fails
authMethod = AuthMethod.SIMPLE;
- sendSaslMessage(outStream, negotiateRequest);
-
+ sendSaslMessage(ipcStreams.out, negotiateRequest);
// loop until sasl is complete or a rpc error occurs
boolean done = false;
do {
- int rpcLen = inStream.readInt();
- ByteBuffer bb = ByteBuffer.allocate(rpcLen);
- inStream.readFully(bb.array());
+ ByteBuffer bb = ipcStreams.readResponse();
RpcWritable.Buffer saslPacket = RpcWritable.Buffer.wrap(bb);
RpcResponseHeaderProto header =
@@ -447,7 +438,7 @@ public class SaslRpcClient {
}
}
if (response != null) {
- sendSaslMessage(outStream, response.build());
+ sendSaslMessage(ipcStreams.out, response.build());
}
} while (!done);
return authMethod;
@@ -461,8 +452,10 @@ public class SaslRpcClient {
ResponseBuffer buf = new ResponseBuffer();
saslHeader.writeDelimitedTo(buf);
message.writeDelimitedTo(buf);
- buf.writeTo(out);
- out.flush();
+ synchronized (out) {
+ buf.writeTo(out);
+ out.flush();
+ }
}
/**
http://git-wip-us.apache.org/repos/asf/hadoop/blob/d4d07687/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml b/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
index 569d894..0a12017 100644
--- a/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
+++ b/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
@@ -1313,10 +1313,19 @@
<name>ipc.maximum.data.length</name>
<value>67108864</value>
<description>This indicates the maximum IPC message length (bytes) that can be
- accepted by the server. Messages larger than this value are rejected by
- server immediately. This setting should rarely need to be changed. It merits
- investigating whether the cause of long RPC messages can be fixed instead,
- e.g. by splitting into smaller messages.
+ accepted by the server. Messages larger than this value are rejected by the
+ immediately to avoid possible OOMs. This setting should rarely need to be
+ changed.
+ </description>
+</property>
+
+<property>
+ <name>ipc.maximum.response.length</name>
+ <value>134217728</value>
+ <description>This indicates the maximum IPC message length (bytes) that can be
+ accepted by the client. Messages larger than this value are rejected
+ immediately to avoid possible OOMs. This setting should rarely need to be
+ changed. Set to 0 to disable.
</description>
</property>
http://git-wip-us.apache.org/repos/asf/hadoop/blob/d4d07687/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPC.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPC.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPC.java
index 82da62d..6fb3511 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPC.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPC.java
@@ -40,6 +40,7 @@ import java.io.OutputStream;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;
+import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.util.ArrayList;
@@ -49,6 +50,8 @@ import java.util.Random;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
@@ -76,6 +79,9 @@ import org.apache.hadoop.ipc.Server.Connection;
import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto;
import org.apache.hadoop.net.ConnectTimeoutException;
import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.SecurityUtil;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.UserGroupInformation.AuthenticationMethod;
import org.apache.hadoop.security.token.SecretManager.InvalidToken;
import org.apache.hadoop.test.GenericTestUtils;
import org.apache.hadoop.util.StringUtils;
@@ -112,6 +118,8 @@ public class TestIPC {
public void setupConf() {
conf = new Configuration();
Client.setPingInterval(conf, PING_INTERVAL);
+ // tests may enable security, so disable before each test
+ UserGroupInformation.setConfiguration(conf);
}
static final Random RANDOM = new Random();
@@ -123,8 +131,8 @@ public class TestIPC {
static ConnectionId getConnectionId(InetSocketAddress addr, int rpcTimeout,
Configuration conf) throws IOException {
- return ConnectionId.getConnectionId(addr, null, null, rpcTimeout, null,
- conf);
+ return ConnectionId.getConnectionId(addr, null,
+ UserGroupInformation.getCurrentUser(), rpcTimeout, null, conf);
}
static Writable call(Client client, InetSocketAddress addr,
@@ -1402,6 +1410,80 @@ public class TestIPC {
client.stop();
}
+ @Test(timeout=4000)
+ public void testInsecureVersionMismatch() throws IOException {
+ checkVersionMismatch();
+ }
+
+ @Test(timeout=4000)
+ public void testSecureVersionMismatch() throws IOException {
+ SecurityUtil.setAuthenticationMethod(AuthenticationMethod.KERBEROS, conf);
+ UserGroupInformation.setConfiguration(conf);
+ checkVersionMismatch();
+ }
+
+ private void checkVersionMismatch() throws IOException {
+ try (final ServerSocket listenSocket = new ServerSocket()) {
+ listenSocket.bind(null);
+ InetSocketAddress addr =
+ (InetSocketAddress) listenSocket.getLocalSocketAddress();
+
+ // open a socket that accepts a client and immediately returns
+ // a version mismatch exception.
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ executor.submit(new Runnable(){
+ @Override
+ public void run() {
+ try {
+ Socket socket = listenSocket.accept();
+ socket.getOutputStream().write(
+ NetworkTraces.RESPONSE_TO_HADOOP_0_20_3_RPC);
+ socket.close();
+ } catch (Throwable t) {
+ // ignore.
+ }
+ }
+ });
+
+ try {
+ Client client = new Client(LongWritable.class, conf);
+ call(client, 0, addr, conf);
+ } catch (RemoteException re) {
+ Assert.assertEquals(RPC.VersionMismatch.class.getName(),
+ re.getClassName());
+ Assert.assertEquals(NetworkTraces.HADOOP0_20_ERROR_MSG,
+ re.getMessage());
+ return;
+ }
+ Assert.fail("didn't get version mismatch");
+ }
+ }
+
+ @Test
+ public void testRpcResponseLimit() throws Throwable {
+ Server server = new TestServer(1, false);
+ InetSocketAddress addr = NetUtils.getConnectAddress(server);
+ server.start();
+
+ conf.setInt(CommonConfigurationKeys.IPC_MAXIMUM_RESPONSE_LENGTH, 0);
+ Client client = new Client(LongWritable.class, conf);
+ call(client, 0, addr, conf);
+
+ conf.setInt(CommonConfigurationKeys.IPC_MAXIMUM_RESPONSE_LENGTH, 4);
+ client = new Client(LongWritable.class, conf);
+ try {
+ call(client, 0, addr, conf);
+ } catch (IOException ioe) {
+ Throwable t = ioe.getCause();
+ Assert.assertNotNull(t);
+ Assert.assertEquals(RpcException.class, t.getClass());
+ Assert.assertEquals("RPC response exceeds maximum data length",
+ t.getMessage());
+ return;
+ }
+ Assert.fail("didn't get limit exceeded");
+ }
+
private void doIpcVersionTest(
byte[] requestData,
byte[] expectedResponse) throws IOException {
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org