GEODE-10675: Refactor membership message read path

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<InternalDistributedMember> messenger;
+  private InternalDistributedMember member;
+
+  @Before
+  @SuppressWarnings("unchecked")
+  public void setUp() throws Exception {
+    ProbeObject.deserialized = false;
+    member = new InternalDistributedMember("localhost", 1234);
+
+    Services<InternalDistributedMember> services = mock(Services.class);
+    when(services.getConfig()).thenReturn(new MembershipConfig() {});
+    when(services.getSerializer()).thenReturn(InternalDataSerializer.getDSFIDSerializer());
+    MemberIdentifierFactory<InternalDistributedMember> memberFactory =
+        mock(MemberIdentifierFactory.class);
+    when(memberFactory.create(any())).thenReturn(member);
+    when(services.getMemberFactory()).thenReturn(memberFactory);
+    JoinLeave<InternalDistributedMember> 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<InternalDistributedMember> result = deserialize(out);
+
+    assertThat(result).isInstanceOf(JoinRequestMessage.class);
+    assertThat(((JoinRequestMessage<InternalDistributedMember>) result).getCredentials())
+        .isEqualTo(credentials);
+  }
+
+  @Test
+  public void deserializeMessageReadsInstallViewWithoutCredentials() throws Exception {
+    HeapDataOutputStream out = startPayload();
+    InternalDataSerializer.writeObject(
+        new InstallViewMessage<>(new GMSMembershipView<>(member), null, false), out);
+
+    Message<InternalDistributedMember> result = deserialize(out);
+
+    assertThat(result).isInstanceOf(InstallViewMessage.class);
+    assertThat(((InstallViewMessage<InternalDistributedMember>) 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<InternalDistributedMember> 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;
+    }
+  }
+}

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<JoinRequestMessage>) message -> messageReceived[0] = message);
+    // configure an incoming message handler for the big message
+    final ByteHolder[] messageReceived = new ByteHolder[1];
+    messenger.addHandler(ByteHolder.class,
+        (MessageHandler<ByteHolder>) 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<MemberIdentifier> {
     static final int DSFID = 123456;
     private byte[] payload;
 

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

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

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();

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<ID> deserializeMessage(DataInputStream in, short ordinal)
     GMSMemberData info = new GMSMemberData();
     info.readEssentialData(in, services.getSerializer().createDeserializationContext(in));
     ID m = services.getMemberFactory().create(info);
-    Message<ID> result = services.getSerializer().getObjectDeserializer().readObject(in);
+    @SuppressWarnings("unchecked")
+    Message<ID> result = (Message<ID>) 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<ID> m) {
     switch (m.getDSFID()) {

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);