[GEODE-10624] Validate interest result policy message part before decoding

geode-core/src/integrationTest/java/org/apache/geode/cache/client/internal/RegisterInterestPolicyPartIntegrationTest.java

@@ -0,0 +1,162 @@
+/*
+ * 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.cache.client.internal;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.Serializable;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.cache.DataPolicy;
+import org.apache.geode.cache.RegionShortcut;
+import org.apache.geode.cache.client.PoolFactory;
+import org.apache.geode.cache.client.PoolManager;
+import org.apache.geode.internal.cache.tier.InterestType;
+import org.apache.geode.internal.cache.tier.MessageType;
+import org.apache.geode.internal.cache.tier.sockets.ChunkedMessage;
+import org.apache.geode.internal.cache.tier.sockets.Message;
+import org.apache.geode.internal.serialization.KnownVersion;
+import org.apache.geode.test.junit.categories.ClientServerTest;
+import org.apache.geode.test.junit.rules.ServerStarterRule;
+
+/**
+ * Exercises, over a real client connection to a running server, how the register-interest command
+ * reads the message part that carries its interest result policy.
+ *
+ * <p>
+ * A client op builds a register-interest request whose policy part carries a type other than the
+ * policy argument, and sends it. The helper type records whether an instance of it is created on
+ * the server while the part is read. The server must read the part only as its expected policy type
+ * and refuse a part carrying any other type.
+ */
+@Category({ClientServerTest.class})
+public class RegisterInterestPolicyPartIntegrationTest {
+
+  private static final String REGION_NAME = "region";
+
+  @Rule
+  public ServerStarterRule server =
+      new ServerStarterRule().withRegion(RegionShortcut.REPLICATE, REGION_NAME).withAutoStart();
+
+  private PoolImpl pool;
+
+  @Before
+  public void setUp() {
+    OtherPartType.reset();
+    final PoolFactory poolFactory = PoolManager.createFactory();
+    poolFactory.addServer("localhost", server.getPort());
+    poolFactory.setReadTimeout(10_000);
+    poolFactory.setMinConnections(1);
+    pool = (PoolImpl) poolFactory.create("testPool");
+  }
+
+  @After
+  public void tearDown() {
+    if (pool != null) {
+      pool.destroy();
+    }
+  }
+
+  @Test
+  public void serverDoesNotProduceAnotherTypeFromThePolicyPart() {
+    try {
+      pool.execute(new PolicyPartOfAnotherTypeOp(REGION_NAME));
+    } catch (final Exception ignored) {
+      // The request does not complete: the point of interest is which type the server produced
+      // while reading the part, which is recorded independently below.
+    }
+
+    assertThat(OtherPartType.instantiated)
+        .as("reading the policy part must not produce a type other than the policy on the server")
+        .isFalse();
+  }
+
+  /**
+   * A register-interest request whose policy part carries a type other than the policy argument.
+   * Sends the request and does not attempt to interpret the response.
+   */
+  private static class PolicyPartOfAnotherTypeOp extends AbstractOp {
+
+    PolicyPartOfAnotherTypeOp(final String region) {
+      super(MessageType.REGISTER_INTEREST, 7);
+      getMessage().addStringPart(region, true);
+      getMessage().addIntPart(InterestType.KEY.ordinal());
+      getMessage().addObjPart(new OtherPartType());
+      getMessage().addBytesPart(new byte[] {(byte) 0x00});
+      getMessage().addStringOrObjPart("key");
+      getMessage().addBytesPart(new byte[] {(byte) 0x00});
+      getMessage().addBytesPart(new byte[] {(byte) DataPolicy.REPLICATE.ordinal(), (byte) 0x01});
+    }
+
+    @Override
+    protected Message createResponseMessage() {
+      return new ChunkedMessage(1, KnownVersion.CURRENT);
+    }
+
+    @Override
+    protected Object processResponse(final Message msg) throws Exception {
+      // Drain the whole response so this op does not return until the server has finished
+      // handling the request.
+      final ChunkedMessage chunkedMessage = (ChunkedMessage) msg;
+      chunkedMessage.readHeader();
+      do {
+        chunkedMessage.receiveChunk();
+      } while (!chunkedMessage.isLastChunk());
+      return null;
+    }
+
+    @Override
+    protected boolean isErrorResponse(final MessageType msgType) {
+      return false;
+    }
+
+    @Override
+    protected long startAttempt(final ConnectionStats stats) {
+      return 0;
+    }
+
+    @Override
+    protected void endSendAttempt(final ConnectionStats stats, final long start) {}
+
+    @Override
+    protected void endAttempt(final ConnectionStats stats, final long start) {}
+  }
+
+  /**
+   * A serializable type other than the register-interest policy argument. It records whether an
+   * instance of it is created, so a test can tell which type a part produced.
+   */
+  public static class OtherPartType implements Serializable {
+    private static final long serialVersionUID = 1L;
+
+    static volatile boolean instantiated = false;
+
+    static void reset() {
+      instantiated = false;
+    }
+
+    private void readObject(final ObjectInputStream in) throws IOException, ClassNotFoundException {
+      in.defaultReadObject();
+      instantiated = true;
+    }
+  }
+}

geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/BaseCommand.java

@@ -82,6 +82,8 @@
 import org.apache.geode.internal.offheap.OffHeapHelper;
 import org.apache.geode.internal.security.SecurityService;
 import org.apache.geode.internal.sequencelog.EntryLogger;
+import org.apache.geode.internal.serialization.DSCODE;
+import org.apache.geode.internal.serialization.DataSerializableFixedID;
 import org.apache.geode.logging.internal.log4j.api.LogService;
 import org.apache.geode.security.GemFireSecurityException;
 import org.apache.geode.util.internal.GeodeGlossary;
@@ -92,6 +94,9 @@ public abstract class BaseCommand implements Command {
   @Immutable
   private static final byte[] OK_BYTES = new byte[] {0};
 
+  /** Length of the serialized form of an interest result policy: code, identifier, ordinal. */
+  private static final int INTEREST_RESULT_POLICY_FORM_LENGTH = 3;
+
   public static final int MAXIMUM_CHUNK_SIZE =
       Integer.getInteger("BridgeServer.MAXIMUM_CHUNK_SIZE", 100);
 
@@ -873,6 +878,39 @@ static Message readRequest(final @NotNull ServerConnection servConn) {
     return requestMsg;
   }
 
+  /**
+   * Reads the interest result policy carried by the given message part.
+   *
+   * <p>
+   * The policy is written by the client in the fixed-identifier form of
+   * {@link InterestResultPolicy}. Only that form is accepted here, so the part is read as a policy
+   * and a part in any other form is refused.
+   *
+   * @param policyPart the message part holding the interest result policy
+   * @return the policy the part describes
+   * @throws IOException if the part is not in the expected form
+   */
+  protected static @NotNull InterestResultPolicy readInterestResultPolicy(
+      final @NotNull Part policyPart) throws IOException, ClassNotFoundException {
+    if (!hasInterestResultPolicyForm(policyPart)) {
+      throw new IOException("The interest result policy part is not in the expected form.");
+    }
+    return (InterestResultPolicy) policyPart.getObject();
+  }
+
+  private static boolean hasInterestResultPolicyForm(final @NotNull Part policyPart) {
+    if (!policyPart.isObject()) {
+      return false;
+    }
+    final byte[] serializedForm = policyPart.getSerializedForm();
+    return serializedForm != null
+        && serializedForm.length == INTEREST_RESULT_POLICY_FORM_LENGTH
+        && serializedForm[0] == DSCODE.DS_FIXED_ID_BYTE.toByte()
+        && serializedForm[1] == DataSerializableFixedID.INTEREST_RESULT_POLICY
+        && serializedForm[2] >= InterestResultPolicy.NONE.getOrdinal()
+        && serializedForm[2] <= InterestResultPolicy.KEYS_VALUES.getOrdinal();
+  }
+
   protected static void fillAndSendRegisterInterestResponseChunks(
       final @Nullable LocalRegion region,
       final @NotNull Object riKey, final @NotNull InterestType interestType,

geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/RegisterInterest61.java

@@ -78,7 +78,7 @@ public void cmdExecute(final @NotNull Message clientMessage,
 
     final InterestResultPolicy policy;
     try {
-      policy = (InterestResultPolicy) clientMessage.getPart(2).getObject();
+      policy = readInterestResultPolicy(clientMessage.getPart(2));
     } catch (Exception e) {
       writeChunkedException(clientMessage, e, serverConnection);
       serverConnection.setAsTrue(RESPONDED);
@@ -113,6 +113,10 @@ public void cmdExecute(final @NotNull Message clientMessage,
     Object key;
     try {
       final Part keyPart = clientMessage.getPart(4);
+      if (interestType == InterestType.REGULAR_EXPRESSION && keyPart.isObject()) {
+        throw new IOException(
+            "The key part of a regular expression request is not in the expected form.");
+      }
       key = keyPart.getStringOrObject();
     } catch (Exception e) {
       writeChunkedException(clientMessage, e, serverConnection);

geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/RegisterInterestList66.java

@@ -79,7 +79,7 @@ public void cmdExecute(final @NotNull Message clientMessage,
     // Retrieve the InterestResultPolicy
     final InterestResultPolicy policy;
     try {
-      policy = (InterestResultPolicy) clientMessage.getPart(1).getObject();
+      policy = readInterestResultPolicy(clientMessage.getPart(1));
     } catch (Exception e) {
       writeChunkedException(clientMessage, e, serverConnection);
       serverConnection.setAsTrue(RESPONDED);

geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/RegisterInterestObjectPartTest.java

@@ -0,0 +1,107 @@
+/*
+ * 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.internal.cache.tier.sockets;
+
+import static org.apache.geode.internal.cache.tier.sockets.BaseCommand.readInterestResultPolicy;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.catchThrowable;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.Serializable;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.cache.InterestResultPolicy;
+import org.apache.geode.internal.util.BlobHelper;
+import org.apache.geode.test.junit.categories.ClientServerTest;
+
+/**
+ * Verifies how the register-interest commands read the message {@link Part} that carries the
+ * interest result policy.
+ *
+ * <p>
+ * The part is read as an {@link InterestResultPolicy}: it is accepted only in the form the client
+ * writes it in, and a part carrying any other type is refused and that type is not produced. The
+ * helper type below records whether an instance of it is created while a part is read.
+ */
+@Category({ClientServerTest.class})
+public class RegisterInterestObjectPartTest {
+
+  @Before
+  public void setUp() {
+    OtherPartType.reset();
+  }
+
+  @Test
+  public void policyPartOfAnotherTypeIsRefusedWithoutProducingThatType() throws Exception {
+    final byte[] objectPartBytes = BlobHelper.serializeToBlob(new OtherPartType());
+
+    final Part part = new Part();
+    part.setPartState(objectPartBytes, true);
+
+    assertThat(catchThrowable(() -> readInterestResultPolicy(part)))
+        .as("a policy part holding another type is refused")
+        .isInstanceOf(IOException.class);
+
+    assertThat(OtherPartType.instantiated)
+        .as("reading the policy part must not produce a type other than the policy")
+        .isFalse();
+  }
+
+  @Test
+  public void nonObjectPolicyPartIsRefused() {
+    final Part part = new Part();
+    part.setPartState(new byte[] {0x01, 0x25, 0x02}, false);
+
+    assertThat(catchThrowable(() -> readInterestResultPolicy(part)))
+        .as("a policy part that is not object typed is refused")
+        .isInstanceOf(IOException.class);
+  }
+
+  @Test
+  public void eachPolicyValueRoundTripsThroughThePart() throws Exception {
+    for (final InterestResultPolicy expected : new InterestResultPolicy[] {
+        InterestResultPolicy.NONE, InterestResultPolicy.KEYS, InterestResultPolicy.KEYS_VALUES}) {
+      final Part part = new Part();
+      part.setPartState(BlobHelper.serializeToBlob(expected), true);
+
+      assertThat(readInterestResultPolicy(part))
+          .as("policy %s survives a write and read of the policy part", expected)
+          .isSameAs(expected);
+    }
+  }
+
+  /**
+   * A serializable type other than the register-interest policy argument. It records whether an
+   * instance of it is created, so a test can tell which type a part produced.
+   */
+  public static class OtherPartType implements Serializable {
+    private static final long serialVersionUID = 1L;
+
+    static volatile boolean instantiated = false;
+
+    static void reset() {
+      instantiated = false;
+    }
+
+    private void readObject(final ObjectInputStream in) throws IOException, ClassNotFoundException {
+      in.defaultReadObject();
+      instantiated = true;
+    }
+  }
+}

geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/command/RegisterInterest61Test.java

@@ -33,6 +33,7 @@
 import org.mockito.MockitoAnnotations;
 
 import org.apache.geode.CancelCriterion;
+import org.apache.geode.cache.InterestResultPolicy;
 import org.apache.geode.cache.operations.RegisterInterestOperationContext;
 import org.apache.geode.internal.cache.InternalCache;
 import org.apache.geode.internal.cache.LocalRegion;
@@ -46,6 +47,7 @@
 import org.apache.geode.internal.security.AuthorizeRequest;
 import org.apache.geode.internal.security.SecurityService;
 import org.apache.geode.internal.serialization.KnownVersion;
+import org.apache.geode.internal.util.BlobHelper;
 import org.apache.geode.security.NotAuthorizedException;
 import org.apache.geode.security.ResourcePermission.Operation;
 import org.apache.geode.security.ResourcePermission.Resource;
@@ -98,6 +100,9 @@ public void setUp() throws Exception {
     when(cache.getRegion(isA(String.class))).thenReturn(uncheckedCast(mock(LocalRegion.class)));
     when(cache.getCancelCriterion()).thenReturn(mock(CancelCriterion.class));
 
+    final Part policyPart = new Part();
+    policyPart.setPartState(BlobHelper.serializeToBlob(InterestResultPolicy.KEYS_VALUES), true);
+
     when(durablePart.getObject()).thenReturn(DURABLE);
 
     when(interestTypePart.getInt()).thenReturn(0);
@@ -107,7 +112,7 @@ public void setUp() throws Exception {
     when(message.getNumberOfParts()).thenReturn(6);
     when(message.getPart(eq(0))).thenReturn(regionNamePart);
     when(message.getPart(eq(1))).thenReturn(interestTypePart);
-    when(message.getPart(eq(2))).thenReturn(mock(Part.class));
+    when(message.getPart(eq(2))).thenReturn(policyPart);
     when(message.getPart(eq(3))).thenReturn(durablePart);
     when(message.getPart(eq(4))).thenReturn(keyPart);
     when(message.getPart(eq(5))).thenReturn(notifyPart);