GEODE-10676: Refactor locator request read path (#8111)
diff --git a/geode-core/src/integrationTest/java/org/apache/geode/distributed/internal/tcpserver/TcpServerRequestFormatIntegrationTest.java b/geode-core/src/integrationTest/java/org/apache/geode/distributed/internal/tcpserver/TcpServerRequestFormatIntegrationTest.java
new file mode 100644
index 0000000..32dbeac
--- /dev/null
+++ b/geode-core/src/integrationTest/java/org/apache/geode/distributed/internal/tcpserver/TcpServerRequestFormatIntegrationTest.java
@@ -0,0 +1,202 @@
+/*
+ * 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.geode.distributed.internal.tcpserver;
+
+import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.Serializable;
+import java.net.InetAddress;
+import java.net.Socket;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.distributed.internal.LocatorRequestDeserializer;
+import org.apache.geode.internal.HeapDataOutputStream;
+import org.apache.geode.internal.InternalDataSerializer;
+import org.apache.geode.internal.serialization.DSFIDSerializer;
+import org.apache.geode.internal.serialization.KnownVersion;
+import org.apache.geode.logging.internal.executors.LoggingExecutors;
+import org.apache.geode.management.internal.configuration.messages.ClusterManagementServiceInfoRequest;
+import org.apache.geode.test.junit.categories.MembershipTest;
+
+/**
+ * Exercises the request-reading path of a locator's {@link TcpServer}, confirming that the type
+ * read
+ * off the request stream is constrained to the locator's request types rather than any type the
+ * stream names, while the real request types are still read.
+ */
+@Category({MembershipTest.class})
+public class TcpServerRequestFormatIntegrationTest {
+
+ private static final int TIMEOUT = 60 * 1000;
+
+ private InetAddress localhost;
+ private TcpServer server;
+ private int port;
+ private DSFIDSerializer serializer;
+ private final AtomicReference<Object> handledRequest = new AtomicReference<>();
+
+ @Before
+ public void setUp() throws IOException {
+ ProbeObject.reconstructed = false;
+ localhost = InetAddress.getLoopbackAddress();
+
+ // Clients encode requests with the general serializer, exactly as a real client does.
+ serializer = InternalDataSerializer.getDSFIDSerializer();
+
+ TcpHandler handler = new TcpHandler() {
+ @Override
+ public Object processRequest(Object request) {
+ handledRequest.set(request);
+ return null;
+ }
+
+ @Override
+ public void endRequest(Object request, long startTime) {}
+
+ @Override
+ public void endResponse(Object request, long startTime) {}
+
+ @Override
+ public void shutDown() {}
+
+ @Override
+ public void init(TcpServer tcpServer) {}
+ };
+
+ // The server reads requests the way a locator does: through the locator request deserializer.
+ server = new TcpServer(0, localhost, handler,
+ "test locator thread",
+ (socket, input, firstByte) -> false,
+ () -> 0,
+ () -> LoggingExecutors.newCachedThreadPool("test thread", true),
+ null,
+ serializer.getObjectSerializer(),
+ new LocatorRequestDeserializer(),
+ "blahblahblah",
+ "blahblahblah");
+ server.start();
+ port = server.getPort();
+ assertThat(port).isGreaterThan(0);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ if (server != null) {
+ TcpClient tcpClient = new TcpClient(new TcpSocketCreatorImpl(),
+ serializer.getObjectSerializer(),
+ serializer.getObjectDeserializer(), TcpSocketFactory.DEFAULT);
+ try {
+ tcpClient.stop(new HostAndPort(localhost.getHostAddress(), port));
+ } catch (Exception ignore) {
+ // best-effort stop
+ }
+ server.join(TIMEOUT);
+ }
+ }
+
+ /**
+ * A request naming a type the locator does not expect is rejected before that type is
+ * reconstructed. {@link ProbeObject} is not a locator request type, so its {@code readObject} is
+ * never run.
+ */
+ @Test
+ public void doesNotReconstructUnexpectedRequestType() throws Exception {
+ sendRequest(new ProbeObject());
+
+ assertThat(ProbeObject.reconstructed).isFalse();
+ assertThat(handledRequest.get()).isNull();
+ }
+
+ /**
+ * A request carried with an explicit class name that the locator does expect is still read. This
+ * covers the data-serialized form.
+ */
+ @Test
+ public void readsExpectedDataSerializableRequest() throws Exception {
+ @SuppressWarnings("deprecation")
+ InfoRequest request = new InfoRequest();
+ sendRequest(request);
+
+ await().untilAsserted(() -> assertThat(handledRequest.get()).isInstanceOf(InfoRequest.class));
+ }
+
+ /**
+ * A request carried in Java-serialized form whose class the locator expects is still read.
+ */
+ @Test
+ public void readsExpectedSerializableRequest() throws Exception {
+ sendRequest(new ClusterManagementServiceInfoRequest());
+
+ await().untilAsserted(
+ () -> assertThat(handledRequest.get())
+ .isInstanceOf(ClusterManagementServiceInfoRequest.class));
+ }
+
+ private void sendRequest(Object request) throws Exception {
+ byte[] encoded = encodeRequest(request);
+ try (Socket socket = new Socket(localhost, port)) {
+ socket.setSoTimeout(TIMEOUT);
+ DataOutputStream out = new DataOutputStream(socket.getOutputStream());
+ out.write(encoded);
+ out.flush();
+ // Drain whatever the server sends (or until it closes) so the request is fully processed.
+ DataInputStream in = new DataInputStream(socket.getInputStream());
+ try {
+ while (in.read() != -1) {
+ // consume
+ }
+ } catch (IOException ignore) {
+ // server closed the socket
+ }
+ }
+ }
+
+ /**
+ * Frames a request the way {@code TcpClient} does: the gossip version int (whose leading byte is
+ * the gossip marker), the serialization version ordinal, then the serialized request object.
+ */
+ private byte[] encodeRequest(Object request) throws IOException {
+ HeapDataOutputStream body = new HeapDataOutputStream(KnownVersion.CURRENT);
+ serializer.getObjectSerializer().writeObject(request, body);
+
+ HeapDataOutputStream out = new HeapDataOutputStream(KnownVersion.CURRENT);
+ out.writeInt(TcpServer.GOSSIPVERSION);
+ out.writeShort(KnownVersion.CURRENT_ORDINAL);
+ out.write(body.toByteArray());
+ return out.toByteArray();
+ }
+
+ /**
+ * Test-only type that records whether it was reconstructed. Not a locator request type.
+ */
+ public static class ProbeObject implements Serializable {
+ static boolean reconstructed;
+
+ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
+ in.defaultReadObject();
+ reconstructed = true;
+ }
+ }
+}
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/InternalLocator.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/InternalLocator.java
index 228617c..6373fef 100644
--- a/geode-core/src/main/java/org/apache/geode/distributed/internal/InternalLocator.java
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/InternalLocator.java
@@ -582,6 +582,7 @@
.setFallbackHandler(handler)
.setLocatorsAreCoordinators(shouldLocatorsBeCoordinators())
.setLocatorStats(locatorStats)
+ .setRequestObjectDeserializer(new LocatorRequestDeserializer())
.create();
} catch (MembershipConfigurationException | UnknownHostException e) {
throw new GemFireConfigException(e.getMessage());
diff --git a/geode-core/src/main/java/org/apache/geode/distributed/internal/LocatorRequestDeserializer.java b/geode-core/src/main/java/org/apache/geode/distributed/internal/LocatorRequestDeserializer.java
new file mode 100644
index 0000000..7e4c3f7
--- /dev/null
+++ b/geode-core/src/main/java/org/apache/geode/distributed/internal/LocatorRequestDeserializer.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.geode.distributed.internal;
+
+import java.io.DataInput;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.geode.annotations.Immutable;
+import org.apache.geode.distributed.internal.tcpserver.InfoRequest;
+import org.apache.geode.distributed.internal.tcpserver.ShutdownRequest;
+import org.apache.geode.distributed.internal.tcpserver.VersionRequest;
+import org.apache.geode.internal.InternalDataSerializer;
+import org.apache.geode.internal.serialization.ObjectDeserializer;
+import org.apache.geode.management.internal.configuration.messages.ClusterManagementServiceInfoRequest;
+import org.apache.geode.management.internal.configuration.messages.SharedConfigurationStatusRequest;
+
+/**
+ * Reads the request a locator receives on its server port. Requests carried as a
+ * {@link org.apache.geode.internal.serialization.DataSerializableFixedID} name a registered fixed
+ * id and are read normally. The remaining request types name their class on the wire; those are
+ * read only for the classes listed here, which is the closed set of locator requests not carried as
+ * a fixed id. A request naming any other class is rejected before that class is loaded.
+ *
+ * <p>
+ * When a new locator request type that is not a {@code DataSerializableFixedID} is added, its class
+ * name must be added to {@link #EXPECTED_REQUEST_CLASS_NAMES}.
+ */
+public class LocatorRequestDeserializer implements ObjectDeserializer {
+
+ @Immutable
+ private static final Set<String> EXPECTED_REQUEST_CLASS_NAMES =
+ Collections.unmodifiableSet(new HashSet<>(Arrays.asList(
+ InfoRequest.class.getName(),
+ ShutdownRequest.class.getName(),
+ VersionRequest.class.getName(),
+ SharedConfigurationStatusRequest.class.getName(),
+ ClusterManagementServiceInfoRequest.class.getName())));
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T> T readObject(DataInput input) throws IOException, ClassNotFoundException {
+ return (T) InternalDataSerializer.readRequestObject(input,
+ EXPECTED_REQUEST_CLASS_NAMES::contains);
+ }
+
+ @Override
+ public void invokeFromData(Object ds, DataInput in) throws IOException, ClassNotFoundException {
+ InternalDataSerializer.invokeFromData(ds, in);
+ }
+}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/InternalDataSerializer.java b/geode-core/src/main/java/org/apache/geode/internal/InternalDataSerializer.java
index d71ea97..69e51e4 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/InternalDataSerializer.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/InternalDataSerializer.java
@@ -23,6 +23,7 @@
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
+import java.io.InvalidClassException;
import java.io.NotSerializableException;
import java.io.ObjectInput;
import java.io.ObjectInputStream;
@@ -67,6 +68,7 @@
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
+import java.util.function.Predicate;
import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.Logger;
@@ -2371,6 +2373,141 @@
}
}
+ /**
+ * Reads an object from a stream that an unauthenticated peer may control, reconstructing only the
+ * types the caller expects. A {@link DataSerializableFixedID} (whose wire form names a registered
+ * fixed id rather than a class) is read normally. An object written with an explicit class name
+ * is
+ * reconstructed only when {@code classIsExpected} accepts that name; for data-serialized objects
+ * the name is checked before the class is loaded. Any other wire form, including a
+ * Java-serialized
+ * object of an unexpected type, is rejected with an {@link IOException}.
+ */
+ public static Object readRequestObject(final DataInput in,
+ final Predicate<String> classIsExpected) throws IOException, ClassNotFoundException {
+ checkIn(in);
+ byte header = in.readByte();
+ DSCODE headerDSCode = DscodeHelper.toDSCODE(header);
+ if (headerDSCode == null) {
+ throw new IOException("Unknown header byte: " + header);
+ }
+ switch (headerDSCode) {
+ case DS_FIXED_ID_BYTE:
+ return dsfidFactory.create(in.readByte(), in);
+ case DS_FIXED_ID_SHORT:
+ return dsfidFactory.create(in.readShort(), in);
+ case DS_FIXED_ID_INT:
+ return dsfidFactory.create(in.readInt(), in);
+ case DS_NO_FIXED_ID:
+ case DATA_SERIALIZABLE:
+ return readDataSerializable(in, classIsExpected);
+ case SERIALIZABLE:
+ return readSerializable(in, classIsExpected);
+ case NULL:
+ case NULL_STRING:
+ return null;
+ default:
+ throw new IOException("Unexpected request format: " + headerDSCode);
+ }
+ }
+
+ private static Object readDataSerializable(final DataInput in,
+ final Predicate<String> classIsExpected) throws IOException, ClassNotFoundException {
+ final Class<?> c = readExpectedClass(in, classIsExpected);
+ try {
+ Constructor<?> init = c.getConstructor();
+ init.setAccessible(true);
+ Object o = init.newInstance();
+
+ invokeFromData(o, in);
+
+ return o;
+ } catch (EOFException | SocketException ex) {
+ // client went away - ignore
+ throw ex;
+ } catch (Exception ex) {
+ throw new SerializationException(
+ String.format("Could not create an instance of %s .",
+ c.getName()),
+ ex);
+ }
+ }
+
+ /**
+ * Reads a class written by {@link DataSerializer#writeClass}, resolving it only when
+ * {@code classIsExpected} accepts its name. The name is checked before the class is loaded, so an
+ * unexpected name is rejected without running that class's static initializer. Primitive type
+ * codes carry no class name and are resolved directly.
+ */
+ private static Class<?> readExpectedClass(final DataInput in,
+ final Predicate<String> classIsExpected) throws IOException, ClassNotFoundException {
+ byte typeCode = in.readByte();
+ if (typeCode == DSCODE.CLASS.toByte()) {
+ String className = processIncomingClassName(readString(in));
+ if (!classIsExpected.test(className)) {
+ throw new IOException("Unexpected class name: " + className);
+ }
+ return getCachedClass(className);
+ }
+ return StaticSerialization.decodePrimitiveClass(typeCode);
+ }
+
+ private static Serializable readSerializable(final DataInput in,
+ final Predicate<String> classIsExpected) throws IOException, ClassNotFoundException {
+ InputStream stream;
+ if (in instanceof InputStream) {
+ stream = (InputStream) in;
+ } else {
+ stream = new InputStream() {
+ @Override
+ public int read() throws IOException {
+ try {
+ return in.readUnsignedByte();
+ } catch (EOFException ignored) {
+ return -1;
+ }
+ }
+ };
+ }
+
+ ObjectInput ois = new ExpectedClassObjectInputStream(stream, classIsExpected);
+
+ if (stream instanceof VersionedDataStream) {
+ KnownVersion v = ((VersionedDataStream) stream).getVersion();
+ if (KnownVersion.CURRENT != v && v != null) {
+ ois = new VersionedObjectInput(ois, v);
+ }
+ }
+
+ return (Serializable) ois.readObject();
+ }
+
+ /**
+ * A {@link DSObjectInputStream} that resolves only classes whose name is accepted by the supplied
+ * predicate. A class outside the accepted set is rejected in {@link #resolveClass}, before it is
+ * loaded, so neither its static initializer nor any instance of it runs.
+ */
+ private static class ExpectedClassObjectInputStream extends DSObjectInputStream {
+
+ private final Predicate<String> classIsExpected;
+
+ ExpectedClassObjectInputStream(InputStream stream, Predicate<String> classIsExpected)
+ throws IOException {
+ super(stream);
+ this.classIsExpected = classIsExpected;
+ }
+
+ @Override
+ protected Class<?> resolveClass(ObjectStreamClass desc)
+ throws IOException, ClassNotFoundException {
+ String className = processIncomingClassName(desc.getName());
+ if (!classIsExpected.test(className)) {
+ throw new InvalidClassException(className, "Unexpected class");
+ }
+ return super.resolveClass(desc);
+ }
+ }
+
public static void writeArrayLength(int len, DataOutput out) throws IOException {
if (len == -1) {
out.writeByte(StaticSerialization.NULL_ARRAY);
diff --git a/geode-membership/src/main/java/org/apache/geode/distributed/internal/membership/api/MembershipLocatorBuilder.java b/geode-membership/src/main/java/org/apache/geode/distributed/internal/membership/api/MembershipLocatorBuilder.java
index 319427c..8945642 100644
--- a/geode-membership/src/main/java/org/apache/geode/distributed/internal/membership/api/MembershipLocatorBuilder.java
+++ b/geode-membership/src/main/java/org/apache/geode/distributed/internal/membership/api/MembershipLocatorBuilder.java
@@ -25,6 +25,7 @@
import org.apache.geode.distributed.internal.tcpserver.TcpHandler;
import org.apache.geode.distributed.internal.tcpserver.TcpSocketCreator;
import org.apache.geode.internal.serialization.DSFIDSerializer;
+import org.apache.geode.internal.serialization.ObjectDeserializer;
public interface MembershipLocatorBuilder<ID extends MemberIdentifier> {
MembershipLocatorBuilder<ID> setPort(int port);
@@ -41,6 +42,14 @@
MembershipLocatorBuilder<ID> setLocatorStats(MembershipLocatorStatistics locatorStats);
+ /**
+ * Sets the deserializer used to read requests received on the locator's server port. When unset,
+ * the serializer's default object deserializer is used. This does not affect the deserializer the
+ * locator uses for its own outgoing requests to other locators.
+ */
+ MembershipLocatorBuilder<ID> setRequestObjectDeserializer(
+ ObjectDeserializer requestObjectDeserializer);
+
MembershipLocator<ID> create()
throws UnknownHostException, MembershipConfigurationException;
diff --git a/geode-membership/src/main/java/org/apache/geode/distributed/internal/membership/gms/MembershipLocatorBuilderImpl.java b/geode-membership/src/main/java/org/apache/geode/distributed/internal/membership/gms/MembershipLocatorBuilderImpl.java
index 4483415..c62f85c 100644
--- a/geode-membership/src/main/java/org/apache/geode/distributed/internal/membership/gms/MembershipLocatorBuilderImpl.java
+++ b/geode-membership/src/main/java/org/apache/geode/distributed/internal/membership/gms/MembershipLocatorBuilderImpl.java
@@ -32,6 +32,7 @@
import org.apache.geode.distributed.internal.tcpserver.TcpSocketCreator;
import org.apache.geode.distributed.internal.tcpserver.TcpSocketCreatorImpl;
import org.apache.geode.internal.serialization.DSFIDSerializer;
+import org.apache.geode.internal.serialization.ObjectDeserializer;
public final class MembershipLocatorBuilderImpl<ID extends MemberIdentifier> implements
MembershipLocatorBuilder<ID> {
@@ -46,6 +47,7 @@
private final Path workingDirectory;
private MembershipConfig config = new MembershipConfig() {};
private final Supplier<ExecutorService> executorServiceSupplier;
+ private ObjectDeserializer requestObjectDeserializer;
public MembershipLocatorBuilderImpl(
final TcpSocketCreator socketCreator,
@@ -101,12 +103,22 @@
}
@Override
+ public MembershipLocatorBuilder<ID> setRequestObjectDeserializer(
+ ObjectDeserializer requestObjectDeserializer) {
+ this.requestObjectDeserializer = requestObjectDeserializer;
+ return this;
+ }
+
+ @Override
public MembershipLocator<ID> create()
throws UnknownHostException, MembershipConfigurationException {
+ ObjectDeserializer serverRequestDeserializer = requestObjectDeserializer == null
+ ? serializer.getObjectDeserializer()
+ : requestObjectDeserializer;
return new MembershipLocatorImpl<>(port, bindAddress, protocolChecker,
executorServiceSupplier,
socketCreator, serializer.getObjectSerializer(), serializer.getObjectDeserializer(),
- fallbackHandler,
+ serverRequestDeserializer, fallbackHandler,
locatorsAreCoordinators, locatorStats, workingDirectory, config);
}
diff --git a/geode-membership/src/main/java/org/apache/geode/distributed/internal/membership/gms/locator/MembershipLocatorImpl.java b/geode-membership/src/main/java/org/apache/geode/distributed/internal/membership/gms/locator/MembershipLocatorImpl.java
index 1edce0e..0381ea4 100644
--- a/geode-membership/src/main/java/org/apache/geode/distributed/internal/membership/gms/locator/MembershipLocatorImpl.java
+++ b/geode-membership/src/main/java/org/apache/geode/distributed/internal/membership/gms/locator/MembershipLocatorImpl.java
@@ -68,6 +68,7 @@
TcpSocketCreator socketCreator,
ObjectSerializer objectSerializer,
ObjectDeserializer objectDeserializer,
+ ObjectDeserializer serverRequestDeserializer,
TcpHandler fallbackHandler,
boolean locatorsAreCoordinators,
MembershipLocatorStatistics locatorStats, Path workingDirectory,
@@ -87,7 +88,7 @@
executorServiceSupplier,
socketCreator,
objectSerializer,
- objectDeserializer,
+ serverRequestDeserializer,
GeodeGlossary.GEMFIRE_PREFIX + "TcpServer.READ_TIMEOUT",
GeodeGlossary.GEMFIRE_PREFIX + "TcpServer.BACKLOG");
