From d5305b90c71b8ec91826a212fb229d43b2517f19 Mon Sep 17 00:00:00 2001 From: JinwooHwang Date: Thu, 3 Sep 2026 19:55:17 -0400 Subject: [PATCH] GEODE-10624: Validate interest result policy message part before decoding Read the register-interest policy part in the fixed-identifier form the client sends it in, and refuse a part in any other form before decoding. Adds unit and integration coverage for the policy part handling. --- ...sterInterestPolicyPartIntegrationTest.java | 162 ++++++++++++++++++ .../cache/tier/sockets/BaseCommand.java | 38 ++++ .../sockets/command/RegisterInterest61.java | 6 +- .../command/RegisterInterestList66.java | 2 +- .../RegisterInterestObjectPartTest.java | 107 ++++++++++++ .../command/RegisterInterest61Test.java | 7 +- .../command/RegisterInterestList66Test.java | 7 +- 7 files changed, 325 insertions(+), 4 deletions(-) create mode 100644 geode-core/src/integrationTest/java/org/apache/geode/cache/client/internal/RegisterInterestPolicyPartIntegrationTest.java create mode 100644 geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/RegisterInterestObjectPartTest.java diff --git a/geode-core/src/integrationTest/java/org/apache/geode/cache/client/internal/RegisterInterestPolicyPartIntegrationTest.java b/geode-core/src/integrationTest/java/org/apache/geode/cache/client/internal/RegisterInterestPolicyPartIntegrationTest.java new file mode 100644 index 00000000000..639e0fd89de --- /dev/null +++ b/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. + * + *

+ * 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; + } + } +} diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/BaseCommand.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/BaseCommand.java index 9000a5503c0..8ca2c7daea8 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/BaseCommand.java +++ b/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. + * + *

+ * 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, diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/RegisterInterest61.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/RegisterInterest61.java index 86984deb164..b8953abf625 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/RegisterInterest61.java +++ b/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); diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/RegisterInterestList66.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/RegisterInterestList66.java index a64197eb80d..c4d06db5ad3 100644 --- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/RegisterInterestList66.java +++ b/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); diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/RegisterInterestObjectPartTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/RegisterInterestObjectPartTest.java new file mode 100644 index 00000000000..b439d0a6f44 --- /dev/null +++ b/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. + * + *

+ * 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; + } + } +} diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/command/RegisterInterest61Test.java b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/command/RegisterInterest61Test.java index 6559f856cef..cfec6dbf98f 100644 --- a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/command/RegisterInterest61Test.java +++ b/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); diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/command/RegisterInterestList66Test.java b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/command/RegisterInterestList66Test.java index a3a6a0f6f50..3d4f1fc34eb 100644 --- a/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/command/RegisterInterestList66Test.java +++ b/geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/command/RegisterInterestList66Test.java @@ -35,6 +35,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; @@ -47,6 +48,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; @@ -103,6 +105,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); @@ -111,7 +116,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(1))).thenReturn(policyPart); when(message.getPart(eq(2))).thenReturn(durablePart); when(message.getPart(eq(3))).thenReturn(keyPart); when(message.getPart(eq(4))).thenReturn(notifyPart);