GEODE-9740: enhance redis security permissions (apache#7029)

Before this commit all geode-for-redis commands required DATA:WRITE.
Now only this commands marked as WRITE commands and the PUBLISH command require DATA:WRITE.
All others now require DATA:READ.

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

@@ -54,11 +54,11 @@ public String getPassword() {
 
   @Override
   protected void setupCacheWithSecurityAndRegionName(String regionName) {
-    setupCacheWithSecurity();
+    setupCacheWithSecurity(false);
   }
 
   @Override
-  public void setupCacheWithSecurity() {
+  public void setupCacheWithSecurity(boolean needsWritePermission) {
     redisContainer =
         new GenericContainer<>(REDIS_DOCKER_IMAGE).withExposedPorts(6379)
             .withCommand("redis-server --requirepass " + getPassword());

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

@@ -50,7 +50,7 @@ public abstract class AbstractAuthIntegrationTest {
 
   protected abstract void setupCacheWithSecurityAndRegionName(String regionName) throws Exception;
 
-  protected abstract void setupCacheWithSecurity() throws Exception;
+  protected abstract void setupCacheWithSecurity(boolean needsWritePermission) throws Exception;
 
   protected abstract void setupCacheWithoutSecurity() throws Exception;
 
@@ -62,7 +62,7 @@ public abstract class AbstractAuthIntegrationTest {
 
   @Test
   public void givenSecurity_authWithIncorrectNumberOfArguments_fails() throws Exception {
-    setupCacheWithSecurity();
+    setupCacheWithSecurity(false);
     assertThatThrownBy(() -> jedis.sendCommand(Protocol.Command.AUTH))
         .hasMessageContaining("ERR wrong number of arguments for 'auth' command");
     assertThatThrownBy(
@@ -72,7 +72,7 @@ public void givenSecurity_authWithIncorrectNumberOfArguments_fails() throws Exce
 
   @Test
   public void givenSecurity_clientCanAuthAfterFailedAuth_passes() throws Exception {
-    setupCacheWithSecurity();
+    setupCacheWithSecurity(false);
 
     assertThatThrownBy(() -> jedis.auth(getUsername(), "wrongpwd"))
         .hasMessageContaining("WRONGPASS invalid username-password pair or user is disabled.");
@@ -83,7 +83,7 @@ public void givenSecurity_clientCanAuthAfterFailedAuth_passes() throws Exception
 
   @Test
   public void givenSecurity_authorizedUser_passes() throws Exception {
-    setupCacheWithSecurity();
+    setupCacheWithSecurity(true);
 
     assertThatThrownBy(() -> jedis.set("foo", "bar"))
         .hasMessage("NOAUTH Authentication required.");
@@ -95,15 +95,15 @@ public void givenSecurity_authorizedUser_passes() throws Exception {
 
   @Test
   public void givenSecurity_authWithCorrectPasswordForDefaultUser_passes() throws Exception {
-    setupCacheWithSecurity();
+    setupCacheWithSecurity(false);
 
     assertThat(jedis.auth(getPassword())).isEqualTo("OK");
     assertThat(jedis.ping()).isEqualTo("PONG");
   }
 
   @Test
   public void givenSecurity_authWithIncorrectPasswordForDefaultUser_fails() throws Exception {
-    setupCacheWithSecurity();
+    setupCacheWithSecurity(false);
 
     assertThatThrownBy(() -> jedis.auth("wrong-password"))
         .hasMessage("WRONGPASS invalid username-password pair or user is disabled.");
@@ -118,7 +118,7 @@ public void givenSecurity_authWithIncorrectPasswordForDefaultUser_fails() throws
    */
   @Test
   public void givenSecurity_separateClientRequest_doNotInteract() throws Exception {
-    setupCacheWithSecurity();
+    setupCacheWithSecurity(true);
     Jedis nonAuthorizedJedis = new Jedis("localhost", getPort(), 100000);
     Jedis authorizedJedis = new Jedis("localhost", getPort(), 100000);
 
@@ -134,7 +134,7 @@ public void givenSecurity_separateClientRequest_doNotInteract() throws Exception
 
   @Test
   public void givenSecurity_lettuceV6AuthClient_passes() throws Exception {
-    setupCacheWithSecurity();
+    setupCacheWithSecurity(false);
 
     RedisURI uri =
         RedisURI.create(String.format("redis://%s@localhost:%d", getUsername(), getPort()));
@@ -160,7 +160,7 @@ public void givenSecurityAndNonDefaultRegionName_authPasses() throws Exception {
 
   @Test
   public void givenSecurity_largeMultiBulkRequestsFail_whenNotAuthenticated() throws Exception {
-    setupCacheWithSecurity();
+    setupCacheWithSecurity(false);
 
     try (Socket clientSocket = new Socket(BIND_ADDRESS, getPort())) {
       clientSocket.setSoTimeout(1000);
@@ -177,7 +177,7 @@ public void givenSecurity_largeMultiBulkRequestsFail_whenNotAuthenticated() thro
 
   @Test
   public void givenSecurity_largeMultiBulkRequestsSucceed_whenAuthenticated() throws Exception {
-    setupCacheWithSecurity();
+    setupCacheWithSecurity(true);
 
     List<String> msetArgs = new ArrayList<>();
     for (int i = 0; i < ByteToCommandDecoder.UNAUTHENTICATED_MAX_ARRAY_SIZE; i++) {
@@ -205,7 +205,7 @@ public void givenNoSecurity_largeMultiBulkRequestsSucceed_whenNotAuthenticated()
 
   @Test
   public void givenSecurity_largeBulkStringRequestsFail_whenNotAuthenticated() throws Exception {
-    setupCacheWithSecurity();
+    setupCacheWithSecurity(false);
 
     try (Socket clientSocket = new Socket(BIND_ADDRESS, getPort())) {
       clientSocket.setSoTimeout(1000);
@@ -222,7 +222,7 @@ public void givenSecurity_largeBulkStringRequestsFail_whenNotAuthenticated() thr
 
   @Test
   public void givenSecurity_largeBulkStringRequestsSucceed_whenAuthenticated() throws Exception {
-    setupCacheWithSecurity();
+    setupCacheWithSecurity(true);
     int stringSize = ByteToCommandDecoder.UNAUTHENTICATED_MAX_BULK_STRING_LENGTH + 1;
 
     String largeString = StringUtils.repeat('a', stringSize);
@@ -244,7 +244,7 @@ public void givenNoSecurity_largeBulkStringRequestsSucceed_whenNotAuthenticated(
 
   @Test
   public void givenSecurity_closingConnectionLogsClientOut() throws Exception {
-    setupCacheWithSecurity();
+    setupCacheWithSecurity(false);
 
     int localPort = AvailablePortHelper.getRandomAvailableTCPPort();
 

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

@@ -45,6 +45,7 @@ public class AuthIntegrationTest extends AbstractAuthIntegrationTest {
   private GeodeRedisServer server;
   private GemFireCache cache;
   private int port;
+  private boolean needsWritePermission;
 
   @Rule
   public RestoreSystemProperties restoreSystemProperties = new RestoreSystemProperties();
@@ -62,20 +63,24 @@ public int getPort() {
 
   @Override
   public String getUsername() {
-    return "dataWrite";
+    if (needsWritePermission) {
+      return "dataWrite";
+    }
+    return "dataRead";
   }
 
   @Override
   public String getPassword() {
-    return "dataWrite";
+    return getUsername();
   }
 
   @Override
   protected void setupCacheWithSecurityAndRegionName(String regionName) throws Exception {
     setupCacheWithRegionName(getUsername(), regionName, true);
   }
 
-  public void setupCacheWithSecurity() throws Exception {
+  public void setupCacheWithSecurity(boolean needsWritePermission) throws Exception {
+    this.needsWritePermission = needsWritePermission;
     setupCache(getUsername(), true);
   }
 
@@ -114,7 +119,7 @@ private void setupCacheWithRegionName(String username, String regionName,
 
   @Test
   public void testAuthConfig() throws Exception {
-    setupCacheWithSecurity();
+    setupCacheWithSecurity(false);
     InternalDistributedSystem iD = (InternalDistributedSystem) cache.getDistributedSystem();
     assertThat(iD.getConfig().getRedisUsername()).isEqualTo(getUsername());
   }
@@ -137,16 +142,45 @@ public void givenNoSecurity_accessWithAuthAndUsernamePassword_fails() throws Exc
 
   @Test
   public void givenSecurity_accessWithCorrectAuthorization_passes() throws Exception {
-    setupCacheWithSecurity();
+    setupCacheWithSecurity(false);
 
     jedis.auth("dataWrite", "dataWrite");
 
     assertThat(jedis.set("foo", "bar")).isEqualTo("OK");
   }
 
+  @Test
+  public void givenSecurity_readOpWithReadAuthorization_passes() throws Exception {
+    setupCacheWithSecurity(false);
+
+    jedis.auth("dataRead", "dataRead");
+
+    assertThat(jedis.get("foo")).isNull();
+  }
+
+  @Test
+  public void givenSecurity_readOpWithWriteAuthorization_fails() throws Exception {
+    setupCacheWithSecurity(false);
+
+    jedis.auth("dataWrite", "dataWrite");
+
+    assertThatThrownBy(() -> jedis.get("foo"))
+        .hasMessageContaining(RedisConstants.ERROR_NOT_AUTHORIZED);
+  }
+
+  @Test
+  public void givenSecurity_writeOpWithReadAuthorization_fails() throws Exception {
+    setupCacheWithSecurity(false);
+
+    jedis.auth("dataRead", "dataRead");
+
+    assertThatThrownBy(() -> jedis.set("foo", "bar"))
+        .hasMessageContaining(RedisConstants.ERROR_NOT_AUTHORIZED);
+  }
+
   @Test
   public void givenSecurity_multipleClientsConnectIndependently() throws Exception {
-    setupCacheWithSecurity();
+    setupCacheWithSecurity(false);
 
     jedis.auth("dataWrite", "dataWrite");
     assertThat(jedis.set("foo", "bar")).isEqualTo("OK");
@@ -159,7 +193,7 @@ public void givenSecurity_multipleClientsConnectIndependently() throws Exception
 
   @Test
   public void givenSecurity_accessWithIncorrectAuthorization_fails() throws Exception {
-    setupCacheWithSecurity();
+    setupCacheWithSecurity(false);
 
     // Authentication is successful
     jedis.auth("dataWriteOther", "dataWriteOther");
@@ -169,4 +203,28 @@ public void givenSecurity_accessWithIncorrectAuthorization_fails() throws Except
         .hasMessageContaining(RedisConstants.ERROR_NOT_AUTHORIZED);
   }
 
+  @Test
+  public void givenSecurityWithReadPermission_clusterCommandSucceeds() throws Exception {
+    setupCacheWithSecurity(false);
+
+    assertThat(jedis.auth(getUsername(), getPassword())).isEqualTo("OK");
+    assertThat(jedis.clusterNodes()).isNotEmpty();
+  }
+
+  @Test
+  public void givenSecurityWithWritePermission_setCommandSucceeds() throws Exception {
+    setupCacheWithSecurity(true);
+
+    assertThat(jedis.auth(getUsername(), getPassword())).isEqualTo("OK");
+    assertThat(jedis.set("foo", "bar")).isEqualTo("OK");
+  }
+
+  @Test
+  public void givenSecurityWithWritePermission_getCommandFails() throws Exception {
+    setupCacheWithSecurity(true);
+
+    assertThat(jedis.auth(getUsername(), getPassword())).isEqualTo("OK");
+    assertThatThrownBy(() -> jedis.get("foo"))
+        .hasMessageContaining(RedisConstants.ERROR_NOT_AUTHORIZED);
+  }
 }

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

@@ -468,6 +468,10 @@ public boolean isAllowedWhileSubscribed() {
     }
   }
 
+  public boolean getRequiresWritePermission() {
+    return parameterRequirements.getFlags().contains(WRITE) || (this == PUBLISH);
+  }
+
   public void checkDeferredParameters(Command command,
       ExecutionHandlerContext executionHandlerContext) {
     deferredParameterRequirements.checkParameters(command, executionHandlerContext);

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

@@ -38,6 +38,7 @@
 import org.apache.shiro.subject.Subject;
 
 import org.apache.geode.CancelException;
+import org.apache.geode.annotations.VisibleForTesting;
 import org.apache.geode.cache.LowMemoryException;
 import org.apache.geode.cache.Region;
 import org.apache.geode.distributed.DistributedMember;
@@ -79,7 +80,15 @@ public class ExecutionHandlerContext extends ChannelInboundHandlerAdapter {
 
   private static final Logger logger = LogService.getLogger();
 
-  private static final ResourcePermission RESOURCE_PERMISSION;
+  private static final ResourcePermission WRITE_RESOURCE_PERMISSION;
+  private static final ResourcePermission READ_RESOURCE_PERMISSION;
+
+  static {
+    String regionName =
+        getStringSystemProperty(REDIS_REGION_NAME_PROPERTY, DEFAULT_REDIS_REGION_NAME);
+    WRITE_RESOURCE_PERMISSION = new ResourcePermission("DATA", "WRITE", regionName);
+    READ_RESOURCE_PERMISSION = new ResourcePermission("DATA", "READ", regionName);
+  }
 
   private final Client client;
   private final RegionProvider regionProvider;
@@ -98,26 +107,6 @@ public class ExecutionHandlerContext extends ChannelInboundHandlerAdapter {
 
   private Subject subject;
 
-  static {
-    ResourcePermission tempPermission;
-    String regionName =
-        getStringSystemProperty(REDIS_REGION_NAME_PROPERTY, DEFAULT_REDIS_REGION_NAME);
-    String resourcePermission = System.getProperty("redis.resource-permission", "DATA:WRITE:"
-        + regionName);
-    String[] parts = resourcePermission.split(":");
-
-    try {
-      if (parts.length != 3) {
-        throw new IllegalArgumentException();
-      }
-      tempPermission = new ResourcePermission(parts[0], parts[1], parts[2]);
-    } catch (IllegalArgumentException e) {
-      tempPermission = new ResourcePermission("DATA", "WRITE", DEFAULT_REDIS_REGION_NAME);
-    }
-
-    RESOURCE_PERMISSION = tempPermission;
-  }
-
   /**
    * Default constructor for execution contexts.
    */
@@ -255,7 +244,7 @@ private void executeCommand(Command command) throws Exception {
         return;
       }
 
-      if (!isAuthorized()) {
+      if (!isAuthorized(command.getCommandType())) {
         writeToChannel(RedisResponse.error(ERROR_NOT_AUTHORIZED));
         return;
       }
@@ -339,13 +328,16 @@ public boolean isAuthenticated() {
     return (!securityService.isEnabled()) || subject != null;
   }
 
-  public boolean isAuthorized() {
-    if (subject == null) {
+  public boolean isAuthorized(RedisCommandType commandType) {
+    if (!securityService.isEnabled()) {
       return true;
     }
 
+    ResourcePermission permission =
+        commandType.getRequiresWritePermission() ? WRITE_RESOURCE_PERMISSION
+            : READ_RESOURCE_PERMISSION;
     try {
-      securityService.authorize(RESOURCE_PERMISSION, subject);
+      securityService.authorize(permission, subject);
     } catch (NotAuthorizedException nex) {
       return false;
     }
@@ -356,7 +348,8 @@ public boolean isAuthorized() {
    * Sets an authenticated principal in the context. This implies that the connection has been
    * successfully authenticated.
    */
-  public void setSubject(Subject subject) {
+  @VisibleForTesting
+  void setSubject(Subject subject) {
     this.subject = subject;
   }