diff --git a/geode-core/src/test/java/org/apache/geode/distributed/internal/membership/gms/messenger/JGroupsMessengerMessageFormatTest.java b/geode-core/src/test/java/org/apache/geode/distributed/internal/membership/gms/messenger/JGroupsMessengerMessageFormatTest.java new file mode 100644 index 000000000000..f35c1415c56d --- /dev/null +++ b/geode-core/src/test/java/org/apache/geode/distributed/internal/membership/gms/messenger/JGroupsMessengerMessageFormatTest.java @@ -0,0 +1,158 @@ +/* + * 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.membership.gms.messenger; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.io.ByteArrayInputStream; +import java.io.DataInputStream; +import java.io.IOException; +import java.io.ObjectOutputStream; +import java.io.Serializable; +import java.util.Properties; + +import org.junit.Before; +import org.junit.Test; + +import org.apache.geode.distributed.internal.membership.InternalDistributedMember; +import org.apache.geode.distributed.internal.membership.api.MemberIdentifierFactory; +import org.apache.geode.distributed.internal.membership.api.MembershipConfig; +import org.apache.geode.distributed.internal.membership.api.Message; +import org.apache.geode.distributed.internal.membership.gms.GMSMembershipView; +import org.apache.geode.distributed.internal.membership.gms.Services; +import org.apache.geode.distributed.internal.membership.gms.interfaces.JoinLeave; +import org.apache.geode.distributed.internal.membership.gms.messages.InstallViewMessage; +import org.apache.geode.distributed.internal.membership.gms.messages.JoinRequestMessage; +import org.apache.geode.internal.HeapDataOutputStream; +import org.apache.geode.internal.InternalDataSerializer; +import org.apache.geode.internal.serialization.DSCODE; +import org.apache.geode.internal.serialization.KnownVersion; + +public class JGroupsMessengerMessageFormatTest { + + private JGroupsMessenger messenger; + private InternalDistributedMember member; + + @Before + @SuppressWarnings("unchecked") + public void setUp() throws Exception { + ProbeObject.deserialized = false; + member = new InternalDistributedMember("localhost", 1234); + + Services services = mock(Services.class); + when(services.getConfig()).thenReturn(new MembershipConfig() {}); + when(services.getSerializer()).thenReturn(InternalDataSerializer.getDSFIDSerializer()); + MemberIdentifierFactory memberFactory = + mock(MemberIdentifierFactory.class); + when(memberFactory.create(any())).thenReturn(member); + when(services.getMemberFactory()).thenReturn(memberFactory); + JoinLeave joinLeave = mock(JoinLeave.class); + when(joinLeave.getMemberID(any())).thenReturn(member); + when(services.getJoinLeave()).thenReturn(joinLeave); + + messenger = new JGroupsMessenger<>(); + messenger.init(services); + } + + @Test + public void deserializeMessageRejectsNonFixedIdPayload() throws Exception { + HeapDataOutputStream out = startPayload(); + out.writeByte(DSCODE.SERIALIZABLE.toByte()); + ObjectOutputStream objectOut = new ObjectOutputStream(out); + objectOut.writeObject(new ProbeObject()); + objectOut.flush(); + + assertThatThrownBy(() -> deserialize(out)).isInstanceOf(IOException.class); + assertThat(ProbeObject.deserialized).isFalse(); + } + + @Test + public void deserializeMessageRejectsJoinRequestWithUnexpectedCredentialsFormat() + throws Exception { + HeapDataOutputStream out = startPayload(); + InternalDataSerializer.writeObject( + new JoinRequestMessage<>(member, member, new ProbeObject(), 0, 0), out); + + assertThatThrownBy(() -> deserialize(out)).isInstanceOf(IOException.class); + assertThat(ProbeObject.deserialized).isFalse(); + } + + @Test + public void deserializeMessageRejectsInstallViewWithUnexpectedCredentialsFormat() + throws Exception { + HeapDataOutputStream out = startPayload(); + InternalDataSerializer.writeObject( + new InstallViewMessage<>(new GMSMembershipView<>(member), new ProbeObject(), false), out); + + assertThatThrownBy(() -> deserialize(out)).isInstanceOf(IOException.class); + assertThat(ProbeObject.deserialized).isFalse(); + } + + @Test + public void deserializeMessageReadsJoinRequestWithPropertiesCredentials() throws Exception { + Properties credentials = new Properties(); + credentials.setProperty("security-username", "user"); + credentials.setProperty("security-password", "secret"); + HeapDataOutputStream out = startPayload(); + InternalDataSerializer.writeObject( + new JoinRequestMessage<>(member, member, credentials, 0, 0), out); + + Message result = deserialize(out); + + assertThat(result).isInstanceOf(JoinRequestMessage.class); + assertThat(((JoinRequestMessage) result).getCredentials()) + .isEqualTo(credentials); + } + + @Test + public void deserializeMessageReadsInstallViewWithoutCredentials() throws Exception { + HeapDataOutputStream out = startPayload(); + InternalDataSerializer.writeObject( + new InstallViewMessage<>(new GMSMembershipView<>(member), null, false), out); + + Message result = deserialize(out); + + assertThat(result).isInstanceOf(InstallViewMessage.class); + assertThat(((InstallViewMessage) result).getCredentials()) + .isNull(); + } + + private HeapDataOutputStream startPayload() throws IOException { + HeapDataOutputStream out = new HeapDataOutputStream(KnownVersion.CURRENT); + member.getMemberData().writeEssentialData(out, + InternalDataSerializer.createSerializationContext(out)); + return out; + } + + private Message deserialize(HeapDataOutputStream out) + throws Exception { + DataInputStream in = new DataInputStream(new ByteArrayInputStream(out.toByteArray())); + return messenger.deserializeMessage(in, KnownVersion.CURRENT_ORDINAL); + } + + public static class ProbeObject implements Serializable { + static boolean deserialized; + + private void readObject(java.io.ObjectInputStream in) + throws IOException, ClassNotFoundException { + in.defaultReadObject(); + deserialized = true; + } + } +} diff --git a/geode-membership/src/integrationTest/java/org/apache/geode/distributed/internal/membership/gms/messenger/JGroupsMessengerJUnitTest.java b/geode-membership/src/integrationTest/java/org/apache/geode/distributed/internal/membership/gms/messenger/JGroupsMessengerJUnitTest.java index d0dc4f42edd0..1901080b7bf9 100755 --- a/geode-membership/src/integrationTest/java/org/apache/geode/distributed/internal/membership/gms/messenger/JGroupsMessengerJUnitTest.java +++ b/geode-membership/src/integrationTest/java/org/apache/geode/distributed/internal/membership/gms/messenger/JGroupsMessengerJUnitTest.java @@ -83,6 +83,7 @@ import org.apache.geode.distributed.internal.membership.gms.interfaces.MessageHandler; import org.apache.geode.distributed.internal.membership.gms.locator.FindCoordinatorRequest; import org.apache.geode.distributed.internal.membership.gms.locator.FindCoordinatorResponse; +import org.apache.geode.distributed.internal.membership.gms.messages.AbstractGMSMessage; import org.apache.geode.distributed.internal.membership.gms.messages.HeartbeatMessage; import org.apache.geode.distributed.internal.membership.gms.messages.InstallViewMessage; import org.apache.geode.distributed.internal.membership.gms.messages.JoinRequestMessage; @@ -518,7 +519,7 @@ public void doTestBigMessageIsFragmented(boolean mcastEnabled, boolean broadcast throws Exception { initMocks(mcastEnabled); MessageHandler mh = mock(MessageHandler.class); - messenger.addHandler(JoinRequestMessage.class, mh); + messenger.addHandler(ByteHolder.class, mh); MemberIdentifier sender = messenger.getMemberID(); GMSMembershipView v = new GMSMembershipView(sender); @@ -528,18 +529,17 @@ public void doTestBigMessageIsFragmented(boolean mcastEnabled, boolean broadcast // send a big message and expect fragmentation MemberIdentifier recipient = broadcastMessage ? null : messenger.localAddress; services.getSerializer().register(ByteHolder.DSFID, ByteHolder.class); - JoinRequestMessage msg = new JoinRequestMessage(recipient, sender, - new ByteHolder( - new byte[(int) (services.getConfig().getUdpFragmentSize() - * 1.5)]), - -1, 0); - + ByteHolder msg = + new ByteHolder(new byte[(int) (services.getConfig().getUdpFragmentSize() * 1.5)]); + if (recipient != null) { + msg.setRecipient(recipient); + } msg.setMulticast(broadcastMessage); - // configure an incoming message handler for JoinRequestMessage - final JoinRequestMessage[] messageReceived = new JoinRequestMessage[1]; - messenger.addHandler(JoinRequestMessage.class, - (MessageHandler) message -> messageReceived[0] = message); + // configure an incoming message handler for the big message + final ByteHolder[] messageReceived = new ByteHolder[1]; + messenger.addHandler(ByteHolder.class, + (MessageHandler) message -> messageReceived[0] = message); // configure the outgoing message interceptor interceptor.mcastSentDataMessages = 0; @@ -1165,7 +1165,7 @@ private MemberIdentifier createAddress(int port) { } - public static class ByteHolder implements DataSerializableFixedID { + public static class ByteHolder extends AbstractGMSMessage { static final int DSFID = 123456; private byte[] payload; diff --git a/geode-membership/src/integrationTest/resources/org/apache/geode/codeAnalysis/sanctionedDataSerializables.txt b/geode-membership/src/integrationTest/resources/org/apache/geode/codeAnalysis/sanctionedDataSerializables.txt index 90624edf119f..24aced9bb854 100644 --- a/geode-membership/src/integrationTest/resources/org/apache/geode/codeAnalysis/sanctionedDataSerializables.txt +++ b/geode-membership/src/integrationTest/resources/org/apache/geode/codeAnalysis/sanctionedDataSerializables.txt @@ -37,11 +37,11 @@ fromData,30 toData,27 org/apache/geode/distributed/internal/membership/gms/messages/InstallViewMessage,2 -fromData,60 +fromData,73 toData,56 org/apache/geode/distributed/internal/membership/gms/messages/JoinRequestMessage,2 -fromData,63 +fromData,76 toData,60 org/apache/geode/distributed/internal/membership/gms/messages/JoinResponseMessage,2 diff --git a/geode-membership/src/main/java/org/apache/geode/distributed/internal/membership/gms/messages/InstallViewMessage.java b/geode-membership/src/main/java/org/apache/geode/distributed/internal/membership/gms/messages/InstallViewMessage.java index f1d012f87d95..0ce66eb3b7be 100755 --- a/geode-membership/src/main/java/org/apache/geode/distributed/internal/membership/gms/messages/InstallViewMessage.java +++ b/geode-membership/src/main/java/org/apache/geode/distributed/internal/membership/gms/messages/InstallViewMessage.java @@ -24,6 +24,7 @@ import org.apache.geode.internal.serialization.DeserializationContext; import org.apache.geode.internal.serialization.KnownVersion; import org.apache.geode.internal.serialization.SerializationContext; +import org.apache.geode.internal.serialization.StaticSerialization; /** * A membership coordinator will send InstallViewMessages to other members when a node @@ -108,7 +109,9 @@ public void fromData(DataInput in, previousViewId = in.readInt(); kind = messageType.values()[in.readInt()]; view = context.getDeserializer().readObject(in); - credentials = context.getDeserializer().readObject(in); + credentials = StaticSerialization.readAllPropertyValueTypes() + ? context.getDeserializer().readObject(in) + : StaticSerialization.readStringPropertiesObject(in); } @Override diff --git a/geode-membership/src/main/java/org/apache/geode/distributed/internal/membership/gms/messages/JoinRequestMessage.java b/geode-membership/src/main/java/org/apache/geode/distributed/internal/membership/gms/messages/JoinRequestMessage.java index 71304b28ecd4..9ba795924e4b 100755 --- a/geode-membership/src/main/java/org/apache/geode/distributed/internal/membership/gms/messages/JoinRequestMessage.java +++ b/geode-membership/src/main/java/org/apache/geode/distributed/internal/membership/gms/messages/JoinRequestMessage.java @@ -23,6 +23,7 @@ import org.apache.geode.internal.serialization.DeserializationContext; import org.apache.geode.internal.serialization.KnownVersion; import org.apache.geode.internal.serialization.SerializationContext; +import org.apache.geode.internal.serialization.StaticSerialization; /** * A JoinRequestMessage is sent from a prospective member of the cluster to a node @@ -114,7 +115,9 @@ public void toData(DataOutput out, public void fromData(DataInput in, DeserializationContext context) throws IOException, ClassNotFoundException { memberID = context.getDeserializer().readObject(in); - credentials = context.getDeserializer().readObject(in); + credentials = StaticSerialization.readAllPropertyValueTypes() + ? context.getDeserializer().readObject(in) + : StaticSerialization.readStringPropertiesObject(in); failureDetectionPort = in.readInt(); // setMulticast(in.readBoolean()); in.readBoolean(); diff --git a/geode-membership/src/main/java/org/apache/geode/distributed/internal/membership/gms/messenger/JGroupsMessenger.java b/geode-membership/src/main/java/org/apache/geode/distributed/internal/membership/gms/messenger/JGroupsMessenger.java index c9df3e225545..8234e731bffb 100644 --- a/geode-membership/src/main/java/org/apache/geode/distributed/internal/membership/gms/messenger/JGroupsMessenger.java +++ b/geode-membership/src/main/java/org/apache/geode/distributed/internal/membership/gms/messenger/JGroupsMessenger.java @@ -92,6 +92,7 @@ import org.apache.geode.distributed.internal.membership.gms.messages.JoinResponseMessage; import org.apache.geode.internal.inet.LocalHostUtil; import org.apache.geode.internal.serialization.BufferDataOutputStream; +import org.apache.geode.internal.serialization.DataSerializableFixedID; import org.apache.geode.internal.serialization.KnownVersion; import org.apache.geode.internal.serialization.StaticSerialization; import org.apache.geode.internal.serialization.VersionedDataInputStream; @@ -1184,13 +1185,28 @@ Message deserializeMessage(DataInputStream in, short ordinal) GMSMemberData info = new GMSMemberData(); info.readEssentialData(in, services.getSerializer().createDeserializationContext(in)); ID m = services.getMemberFactory().create(info); - Message result = services.getSerializer().getObjectDeserializer().readObject(in); + @SuppressWarnings("unchecked") + Message result = (Message) readFixedIdObject(in); setSender(result, m, ordinal); return result; } + /** + * Reads the message body, which is always written as a {@link DataSerializableFixedID}. A payload + * whose leading type code is anything else is rejected with an {@link IOException}. + */ + private Object readFixedIdObject(DataInputStream in) throws IOException, ClassNotFoundException { + int dsfid; + try { + dsfid = services.getSerializer().readDSFIDHeader(in); + } catch (IllegalStateException e) { + throw new IOException("Unexpected message format", e); + } + return services.getSerializer().create(dsfid, in); + } + /** look for certain messages that may need to be altered before being sent */ void filterOutgoingMessage(Message m) { switch (m.getDSFID()) { diff --git a/geode-serialization/src/main/java/org/apache/geode/internal/serialization/StaticSerialization.java b/geode-serialization/src/main/java/org/apache/geode/internal/serialization/StaticSerialization.java index 1ffbb0f65870..318aba7e4ff1 100644 --- a/geode-serialization/src/main/java/org/apache/geode/internal/serialization/StaticSerialization.java +++ b/geode-serialization/src/main/java/org/apache/geode/internal/serialization/StaticSerialization.java @@ -21,8 +21,10 @@ import java.net.UnknownHostException; import java.util.HashMap; import java.util.Map; +import java.util.Properties; import org.apache.geode.annotations.internal.MakeNotStatic; +import org.apache.geode.util.internal.GeodeGlossary; /** * StaticSerialization provides a collection of serialization/deserialization methods that @@ -217,6 +219,77 @@ private static String readHugeStringFromDataInput(DataInput in) throws IOExcepti return new String(buf); } + @MakeNotStatic + private static final boolean READ_ALL_PROPERTY_VALUE_TYPES = + Boolean.getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "readAllPropertyValueTypes"); + + /** + * Whether property values should be read with the general object reader rather than the simple + * reader used by {@link #readStringProperties}. Set the {@code gemfire.readAllPropertyValueTypes} + * system property for deployments that place value types other than strings and byte arrays in + * their connection properties. + */ + public static boolean readAllPropertyValueTypes() { + return READ_ALL_PROPERTY_VALUE_TYPES; + } + + /** + * Reads the body of a {@link Properties} written with the standard properties encoding, accepting + * only the string, byte-array and null forms expected for connection properties. An entry encoded + * in any other form is rejected with an {@link IOException}. + */ + public static Properties readStringProperties(DataInput in) throws IOException { + int size = readArrayLength(in); + if (size == -1) { + return null; + } + Properties props = new Properties(); + for (int index = 0; index < size; index++) { + Object key = readSimplePropertyValue(in); + Object value = readSimplePropertyValue(in); + props.put(key, value); + } + return props; + } + + /** + * Reads an object-encoded {@link Properties} (a leading type code followed by the body), + * accepting + * only the null and properties forms and restricting values as {@link #readStringProperties} + * does. Any other leading type code is rejected with an {@link IOException}. + */ + public static Object readStringPropertiesObject(DataInput in) throws IOException { + DSCODE dscode = DscodeHelper.toDSCODE(in.readByte()); + if (dscode == DSCODE.NULL) { + return null; + } + if (dscode == DSCODE.PROPERTIES) { + return readStringProperties(in); + } + throw new IOException("Unexpected wire format while reading properties"); + } + + private static Object readSimplePropertyValue(DataInput in) throws IOException { + DSCODE dscode = DscodeHelper.toDSCODE(in.readByte()); + if (dscode == null) { + throw new IOException("Unexpected wire format while reading properties"); + } + switch (dscode) { + case NULL: + case NULL_STRING: + return null; + case STRING: + case HUGE_STRING: + case STRING_BYTES: + case HUGE_STRING_BYTES: + return readString(in, dscode); + case BYTE_ARRAY: + return readByteArray(in); + default: + throw new IOException("Unexpected wire format while reading properties"); + } + } + public static String[] readStringArray(DataInput in) throws IOException { int length = readArrayLength(in);