From 5c92a5e1acce42be1f8103d510ce2ad941ea7f07 Mon Sep 17 00:00:00 2001 From: JinwooHwang Date: Wed, 14 Oct 2026 13:17:48 -0400 Subject: [PATCH] GEODE-10676: Refactor locator request read path --- ...TcpServerRequestFormatIntegrationTest.java | 202 ++++++++++++++++++ .../distributed/internal/InternalLocator.java | 1 + .../internal/LocatorRequestDeserializer.java | 66 ++++++ .../internal/InternalDataSerializer.java | 137 ++++++++++++ .../api/MembershipLocatorBuilder.java | 9 + .../gms/MembershipLocatorBuilderImpl.java | 14 +- .../gms/locator/MembershipLocatorImpl.java | 3 +- 7 files changed, 430 insertions(+), 2 deletions(-) create mode 100644 geode-core/src/integrationTest/java/org/apache/geode/distributed/internal/tcpserver/TcpServerRequestFormatIntegrationTest.java create mode 100644 geode-core/src/main/java/org/apache/geode/distributed/internal/LocatorRequestDeserializer.java 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 000000000000..32dbeacee525 --- /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 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 228617c4a509..6373fef2e930 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 @@ public static boolean isDedicatedLocator() { .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 000000000000..7e4c3f710a03 --- /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. + * + *

+ * 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 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 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 5b785e87b430..75717352a930 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 @@ private static Object readDataSerializable(final DataInput in) } } + /** + * 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 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 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 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 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 classIsExpected; + + ExpectedClassObjectInputStream(InputStream stream, Predicate 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 319427c46156..89456426edc1 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 { MembershipLocatorBuilder setPort(int port); @@ -41,6 +42,14 @@ public interface MembershipLocatorBuilder { MembershipLocatorBuilder 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 setRequestObjectDeserializer( + ObjectDeserializer requestObjectDeserializer); + MembershipLocator 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 44834152b281..c62f85ce72b5 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 implements MembershipLocatorBuilder { @@ -46,6 +47,7 @@ public final class MembershipLocatorBuilderImpl imp private final Path workingDirectory; private MembershipConfig config = new MembershipConfig() {}; private final Supplier executorServiceSupplier; + private ObjectDeserializer requestObjectDeserializer; public MembershipLocatorBuilderImpl( final TcpSocketCreator socketCreator, @@ -100,13 +102,23 @@ public MembershipLocatorBuilder setLocatorStats(MembershipLocatorStatistics return this; } + @Override + public MembershipLocatorBuilder setRequestObjectDeserializer( + ObjectDeserializer requestObjectDeserializer) { + this.requestObjectDeserializer = requestObjectDeserializer; + return this; + } + @Override public MembershipLocator 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 1edce0ebd0a1..0381ea4f445f 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 @@ public MembershipLocatorImpl(int port, HostAddress bindAddress, TcpSocketCreator socketCreator, ObjectSerializer objectSerializer, ObjectDeserializer objectDeserializer, + ObjectDeserializer serverRequestDeserializer, TcpHandler fallbackHandler, boolean locatorsAreCoordinators, MembershipLocatorStatistics locatorStats, Path workingDirectory, @@ -87,7 +88,7 @@ public MembershipLocatorImpl(int port, HostAddress bindAddress, executorServiceSupplier, socketCreator, objectSerializer, - objectDeserializer, + serverRequestDeserializer, GeodeGlossary.GEMFIRE_PREFIX + "TcpServer.READ_TIMEOUT", GeodeGlossary.GEMFIRE_PREFIX + "TcpServer.BACKLOG");