GEODE-9676: Limit array and string sizes for unauthenticated Radish connections

geode-for-redis/src/acceptanceTest/java/org/apache/geode/redis/internal/executor/connection/AuthNativeRedisAcceptanceTest.java

@@ -27,7 +27,7 @@
 
 public class AuthNativeRedisAcceptanceTest extends AbstractAuthIntegrationTest {
 
-  private static final String REDIS_DOCKER_IMAGE = "redis:6.2.4";
+  private static final String REDIS_DOCKER_IMAGE = "redis:6.2.6";
 
   // Docker compose does not work on windows in CI. Ignore this test on windows
   // Using a RuleChain to make sure we ignore the test before the rule comes into play

geode-for-redis/src/acceptanceTest/resources/0001-configure-redis-tests.patch

@@ -74,7 +74,7 @@ index 8978631e0..33c10eb15 100644
      } elseif {$opt eq {--port}} {
          set ::port $arg
 diff --git a/tests/unit/auth.tcl b/tests/unit/auth.tcl
-index f5da728e8..6ae96f295 100644
+index f5da728e8..13985dce2 100644
 --- a/tests/unit/auth.tcl
 +++ b/tests/unit/auth.tcl
 @@ -1,15 +1,16 @@
@@ -110,41 +110,6 @@ index f5da728e8..6ae96f295 100644
      } {OK}
 
      test {Once AUTH succeeded we can actually send commands to the server} {
-@@ -25,19 +26,19 @@ start_server {tags {"auth"} overrides {requirepass foobar}} {
-         r incr foo
-     } {101}
-
--    test {For unauthenticated clients multibulk and bulk length are limited} {
--        set rr [redis [srv "host"] [srv "port"] 0]
--        $rr write "*100\r\n"
--        $rr flush
--        catch {[$rr read]} e
--        assert_match {*unauthenticated multibulk length*} $e
--        $rr close
--
--        set rr [redis [srv "host"] [srv "port"] 0]
--        $rr write "*1\r\n\$100000000\r\n"
--        $rr flush
--        catch {[$rr read]} e
--        assert_match {*unauthenticated bulk length*} $e
--        $rr close
--    }
-+#    test {For unauthenticated clients multibulk and bulk length are limited} {
-+#        set rr [redis [srv "host"] [srv "port"] 0]
-+#        $rr write "*100\r\n"
-+#        $rr flush
-+#        catch {[$rr read]} e
-+#        assert_match {*unauthenticated multibulk length*} $e
-+#        $rr close
-+#
-+#        set rr [redis [srv "host"] [srv "port"] 0]
-+#        $rr write "*1\r\n\$100000000\r\n"
-+#        $rr flush
-+#        catch {[$rr read]} e
-+#        assert_match {*unauthenticated bulk length*} $e
-+#        $rr close
-+#    }
- }
 diff --git a/tests/unit/dump.tcl b/tests/unit/dump.tcl
 index 4c4e5d075..18bb694f2 100644
 --- a/tests/unit/dump.tcl

geode-for-redis/src/integrationTest/java/org/apache/geode/redis/internal/executor/connection/AbstractAuthIntegrationTest.java

@@ -14,15 +14,35 @@
  */
 package org.apache.geode.redis.internal.executor.connection;
 
+import static org.apache.geode.redis.internal.RedisConstants.ERROR_NOT_AUTHENTICATED;
+import static org.apache.geode.redis.internal.RedisConstants.ERROR_UNAUTHENTICATED_BULK;
+import static org.apache.geode.redis.internal.RedisConstants.ERROR_UNAUTHENTICATED_MULTIBULK;
+import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
+import static org.apache.geode.test.dunit.rules.RedisClusterStartupRule.BIND_ADDRESS;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNoException;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
+import java.io.PrintWriter;
+import java.net.InetAddress;
+import java.net.Socket;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
+
 import io.lettuce.core.RedisClient;
 import io.lettuce.core.RedisURI;
+import org.apache.commons.lang3.StringUtils;
 import org.junit.Test;
 import redis.clients.jedis.Jedis;
 import redis.clients.jedis.Protocol;
 
+import org.apache.geode.internal.AvailablePortHelper;
+import org.apache.geode.redis.internal.netty.ByteToCommandDecoder;
+
 
 public abstract class AbstractAuthIntegrationTest {
 
@@ -128,4 +148,128 @@ public void givenNoSecurity_accessWithoutAuth_passes() throws Exception {
     assertThat(jedis.ping()).isEqualTo("PONG");
   }
 
+  @Test
+  public void givenSecurity_largeMultiBulkRequestsFail_whenNotAuthenticated() throws Exception {
+    setupCacheWithSecurity();
+
+    try (Socket clientSocket = new Socket(BIND_ADDRESS, getPort())) {
+      clientSocket.setSoTimeout(1000);
+      PrintWriter out = new PrintWriter(clientSocket.getOutputStream());
+      BufferedReader in = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
+
+      out.write("*100\r\n");
+      out.flush();
+      String response = in.readLine();
+
+      assertThat(response).contains(ERROR_UNAUTHENTICATED_MULTIBULK);
+    }
+  }
+
+  @Test
+  public void givenSecurity_largeMultiBulkRequestsSucceed_whenAuthenticated() throws Exception {
+    setupCacheWithSecurity();
+
+    List<String> msetArgs = new ArrayList<>();
+    for (int i = 0; i < ByteToCommandDecoder.UNAUTHENTICATED_MAX_ARRAY_SIZE; i++) {
+      msetArgs.add("{hash}key-" + i);
+      msetArgs.add("value-" + i);
+    }
+
+    assertThat(jedis.auth(getUsername(), getPassword())).isEqualTo("OK");
+    assertThat(jedis.mset(msetArgs.toArray(new String[] {}))).isEqualTo("OK");
+  }
+
+  @Test
+  public void givenNoSecurity_largeMultiBulkRequestsSucceed_whenNotAuthenticated()
+      throws Exception {
+    setupCacheWithoutSecurity();
+
+    List<String> msetArgs = new ArrayList<>();
+    for (int i = 0; i < ByteToCommandDecoder.UNAUTHENTICATED_MAX_ARRAY_SIZE; i++) {
+      msetArgs.add("{hash}key-" + i);
+      msetArgs.add("value-" + i);
+    }
+
+    assertThat(jedis.mset(msetArgs.toArray(new String[] {}))).isEqualTo("OK");
+  }
+
+  @Test
+  public void givenSecurity_largeBulkStringRequestsFail_whenNotAuthenticated() throws Exception {
+    setupCacheWithSecurity();
+
+    try (Socket clientSocket = new Socket(BIND_ADDRESS, getPort())) {
+      clientSocket.setSoTimeout(1000);
+      PrintWriter out = new PrintWriter(clientSocket.getOutputStream());
+      BufferedReader in = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
+
+      out.write("*1\r\n$100000000\r\n");
+      out.flush();
+      String response = in.readLine();
+
+      assertThat(response).contains(ERROR_UNAUTHENTICATED_BULK);
+    }
+  }
+
+  @Test
+  public void givenSecurity_largeBulkStringRequestsSucceed_whenAuthenticated() throws Exception {
+    setupCacheWithSecurity();
+    int stringSize = ByteToCommandDecoder.UNAUTHENTICATED_MAX_BULK_STRING_LENGTH + 1;
+
+    String largeString = StringUtils.repeat('a', stringSize);
+
+    assertThat(jedis.auth(getUsername(), getPassword())).isEqualTo("OK");
+    assertThat(jedis.set("key", largeString)).isEqualTo("OK");
+  }
+
+  @Test
+  public void givenNoSecurity_largeBulkStringRequestsSucceed_whenNotAuthenticated()
+      throws Exception {
+    setupCacheWithoutSecurity();
+    int stringSize = ByteToCommandDecoder.UNAUTHENTICATED_MAX_BULK_STRING_LENGTH + 1;
+
+    String largeString = StringUtils.repeat('a', stringSize);
+
+    assertThat(jedis.set("key", largeString)).isEqualTo("OK");
+  }
+
+  @Test
+  public void givenSecurity_closingConnectionLogsClientOut() throws Exception {
+    setupCacheWithSecurity();
+
+    int localPort = AvailablePortHelper.getRandomAvailableTCPPort();
+
+    try (Socket clientSocket = new Socket(BIND_ADDRESS, getPort(), InetAddress.getLoopbackAddress(),
+        localPort)) {
+      clientSocket.setSoTimeout(1000);
+      PrintWriter out = new PrintWriter(clientSocket.getOutputStream());
+      BufferedReader in = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
+
+      out.write("*3\r\n$4\r\nAUTH\r\n" +
+          "$" + getUsername().length() + "\r\n" + getUsername() + "\r\n" +
+          "$" + getPassword().length() + "\r\n" + getPassword() + "\r\n");
+      out.flush();
+      String response = in.readLine();
+
+      assertThat(response).contains("OK");
+    }
+
+    AtomicReference<Socket> socketRef = new AtomicReference<>(null);
+
+    await().pollInterval(Duration.ofSeconds(1))
+        .untilAsserted(() -> assertThatNoException().isThrownBy(() -> socketRef.set(
+            new Socket(BIND_ADDRESS, getPort(), InetAddress.getLoopbackAddress(), localPort))));
+
+    try (Socket clientSocket = socketRef.get()) {
+      clientSocket.setSoTimeout(1000);
+      PrintWriter out = new PrintWriter(clientSocket.getOutputStream());
+      BufferedReader in = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
+
+      out.write("*1\r\n$4\r\nPING\r\n");
+      out.flush();
+      String response = in.readLine();
+
+      assertThat(response).contains(ERROR_NOT_AUTHENTICATED);
+    }
+  }
+
 }

geode-for-redis/src/main/java/org/apache/geode/redis/internal/GeodeRedisServer.java

@@ -22,7 +22,6 @@
 import org.apache.geode.distributed.internal.InternalDistributedSystem;
 import org.apache.geode.internal.cache.InternalCache;
 import org.apache.geode.internal.cache.PartitionedRegion;
-import org.apache.geode.internal.security.SecurityService;
 import org.apache.geode.internal.statistics.StatisticsClock;
 import org.apache.geode.internal.statistics.StatisticsClockFactory;
 import org.apache.geode.logging.internal.log4j.api.LogService;
@@ -35,6 +34,7 @@
 import org.apache.geode.redis.internal.pubsub.PubSubImpl;
 import org.apache.geode.redis.internal.pubsub.Subscriptions;
 import org.apache.geode.redis.internal.services.LockingStripedCoordinator;
+import org.apache.geode.redis.internal.services.RedisSecurityService;
 import org.apache.geode.redis.internal.services.StripedCoordinator;
 import org.apache.geode.redis.internal.statistics.GeodeRedisStats;
 import org.apache.geode.redis.internal.statistics.RedisStats;
@@ -84,7 +84,7 @@ public GeodeRedisServer(String bindAddress, int port, InternalCache cache) {
     passiveExpirationManager = new PassiveExpirationManager(regionProvider);
 
     DistributedMember member = cache.getDistributedSystem().getDistributedMember();
-    SecurityService securityService = cache.getSecurityService();
+    RedisSecurityService securityService = new RedisSecurityService(cache.getSecurityService());
 
     nettyRedisServer = new NettyRedisServer(() -> cache.getInternalDistributedSystem().getConfig(),
         regionProvider, pubSub,

geode-for-redis/src/main/java/org/apache/geode/redis/internal/RedisConstants.java

@@ -71,5 +71,9 @@ public class RedisConstants {
       "AUTH called without a Security Manager configured.";
   public static final String ERROR_KEY_REQUIRED =
       "at least 1 input key is needed for ZUNIONSTORE/ZINTERSTORE";
+  public static final String ERROR_UNAUTHENTICATED_MULTIBULK =
+      "Protocol error: unauthenticated multibulk length";
+  public static final String ERROR_UNAUTHENTICATED_BULK =
+      "Protocol error: unauthenticated bulk length";
   public static final String INTERNAL_SERVER_ERROR = "Internal server error: ";
 }

geode-for-redis/src/main/java/org/apache/geode/redis/internal/executor/connection/AuthExecutor.java

@@ -22,9 +22,6 @@
 import java.util.List;
 import java.util.Properties;
 
-import org.apache.shiro.subject.Subject;
-
-import org.apache.geode.internal.security.SecurityService;
 import org.apache.geode.redis.internal.executor.CommandExecutor;
 import org.apache.geode.redis.internal.executor.RedisResponse;
 import org.apache.geode.redis.internal.netty.Command;
@@ -37,18 +34,15 @@ public class AuthExecutor implements CommandExecutor {
 
   @Override
   public RedisResponse executeCommand(Command command, ExecutionHandlerContext context) {
-    SecurityService securityService = context.getSecurityService();
-
     // We're deviating from Redis here in that any AUTH requests, without security explicitly
     // set up, will fail.
-    if (!securityService.isIntegratedSecurity()) {
+    if (!context.isSecurityEnabled()) {
       return RedisResponse.error(ERROR_AUTH_CALLED_WITHOUT_SECURITY_CONFIGURED);
     }
 
     Properties props = getSecurityProperties(command, context);
     try {
-      Subject subject = securityService.login(props);
-      context.setSubject(subject);
+      context.login(props);
     } catch (AuthenticationFailedException | AuthenticationExpiredException ex) {
       return RedisResponse.wrongpass(ERROR_INVALID_USERNAME_OR_PASSWORD);
     }

geode-for-redis/src/main/java/org/apache/geode/redis/internal/netty/ByteToCommandDecoder.java

@@ -15,6 +15,8 @@
  */
 package org.apache.geode.redis.internal.netty;
 
+import static org.apache.geode.redis.internal.RedisConstants.ERROR_UNAUTHENTICATED_BULK;
+import static org.apache.geode.redis.internal.RedisConstants.ERROR_UNAUTHENTICATED_MULTIBULK;
 import static org.apache.geode.redis.internal.netty.StringBytesGlossary.ARRAY_ID;
 import static org.apache.geode.redis.internal.netty.StringBytesGlossary.BULK_STRING_ID;
 import static org.apache.geode.redis.internal.netty.StringBytesGlossary.bCRLF;
@@ -25,8 +27,11 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelId;
 import io.netty.handler.codec.ByteToMessageDecoder;
 
+import org.apache.geode.redis.internal.RedisException;
+import org.apache.geode.redis.internal.services.RedisSecurityService;
 import org.apache.geode.redis.internal.statistics.RedisStats;
 
 /**
@@ -45,12 +50,27 @@
  */
 public class ByteToCommandDecoder extends ByteToMessageDecoder {
 
+  public static final String UNAUTHENTICATED_MAX_ARRAY_SIZE_PARAM =
+      "gemfire.geode-for-redis-unauthenticated-max-array-size";
+  public static final String UNAUTHENTICATED_MAX_BULK_STRING_LENGTH_PARAM =
+      "gemfire.geode-for-redis-unauthenticated-max-bulk-string-length";
+
   private static final int MAX_BULK_STRING_LENGTH = 512 * 1024 * 1024; // 512 MB
+  // These 2 defaults are taken from native Redis
+  public static final int UNAUTHENTICATED_MAX_ARRAY_SIZE =
+      Integer.getInteger(UNAUTHENTICATED_MAX_ARRAY_SIZE_PARAM, 10);
+  public static final int UNAUTHENTICATED_MAX_BULK_STRING_LENGTH =
+      Integer.getInteger(UNAUTHENTICATED_MAX_BULK_STRING_LENGTH_PARAM, 16384);
 
   private final RedisStats redisStats;
+  private final RedisSecurityService securityService;
+  private final ChannelId channelId;
 
-  public ByteToCommandDecoder(RedisStats redisStats) {
+  public ByteToCommandDecoder(RedisStats redisStats, RedisSecurityService securityService,
+      ChannelId channelId) {
     this.redisStats = redisStats;
+    this.securityService = securityService;
+    this.channelId = channelId;
   }
 
   @Override
@@ -96,11 +116,13 @@ private List<byte[]> parseArray(ByteBuf buffer)
       throws RedisCommandParserException {
     byte currentChar;
     int arrayLength = parseCurrentNumber(buffer);
-    if (arrayLength == Integer.MIN_VALUE || !parseRN(buffer)) {
+    if (arrayLength < 0 || !parseRN(buffer)) {
       return null;
     }
-    if (arrayLength < 0 || arrayLength > 1000000000) {
-      throw new RedisCommandParserException("invalid multibulk length");
+
+    if (arrayLength > UNAUTHENTICATED_MAX_ARRAY_SIZE
+        && securityService.isEnabled() && !securityService.isAuthenticated(channelId)) {
+      throw new RedisException(ERROR_UNAUTHENTICATED_MULTIBULK);
     }
 
     List<byte[]> commandElems = new ArrayList<>(arrayLength);
@@ -133,13 +155,20 @@ private List<byte[]> parseArray(ByteBuf buffer)
    */
   private byte[] parseBulkString(ByteBuf buffer) throws RedisCommandParserException {
     int bulkStringLength = parseCurrentNumber(buffer);
-    if (bulkStringLength == Integer.MIN_VALUE) {
+    if (bulkStringLength < 0) {
       return null;
     }
+
     if (bulkStringLength > MAX_BULK_STRING_LENGTH) {
       throw new RedisCommandParserException(
           "invalid bulk length, cannot exceed max length of " + MAX_BULK_STRING_LENGTH);
     }
+
+    if (bulkStringLength > UNAUTHENTICATED_MAX_BULK_STRING_LENGTH
+        && securityService.isEnabled() && !securityService.isAuthenticated(channelId)) {
+      throw new RedisException(ERROR_UNAUTHENTICATED_BULK);
+    }
+
     if (!parseRN(buffer)) {
       return null;
     }