Merge pull request #2418 from Git-Yang/enhanced_acl

[ISSUE #2328] Add parameter validation to ACL

acl/src/main/java/org/apache/rocketmq/acl/common/AclConstants.java

@@ -44,6 +44,16 @@ public class AclConstants {
 
     public static final String CONFIG_TIME_STAMP = "timestamp";
 
+    public static final String PUB = "PUB";
+
+    public static final String SUB = "SUB";
+
+    public static final String DENY = "DENY";
+
+    public static final String PUB_SUB = "PUB|SUB";
+
+    public static final String SUB_PUB = "SUB|PUB";
+
     public static final int ACCESS_KEY_MIN_LENGTH = 6;
 
     public static final int SECRET_KEY_MIN_LENGTH = 6;

acl/src/main/java/org/apache/rocketmq/acl/common/Permission.java

@@ -60,14 +60,14 @@ public class Permission {
             return Permission.DENY;
         }
         switch (permString.trim()) {
-            case "PUB":
+            case AclConstants.PUB:
                 return Permission.PUB;
-            case "SUB":
+            case AclConstants.SUB:
                 return Permission.SUB;
-            case "PUB|SUB":
-            case "SUB|PUB":
+            case AclConstants.PUB_SUB:
+            case AclConstants.SUB_PUB:
                 return Permission.PUB | Permission.SUB;
-            case "DENY":
+            case AclConstants.DENY:
                 return Permission.DENY;
             default:
                 return Permission.DENY;
@@ -89,6 +89,25 @@ public class Permission {
         }
     }
 
+    public static void checkResourcePerms(List<String> resources) {
+        if (resources == null || resources.isEmpty()) {
+            return;
+        }
+
+        for (String resource : resources) {
+            String[] items = StringUtils.split(resource, "=");
+            if (items.length != 2) {
+                throw new AclException(String.format("Parse Resource format error for %s.\n" +
+                    "The expected resource format is 'Res=Perm'. For example: topicA=SUB", resource));
+            }
+
+            if (!AclConstants.DENY.equals(items[1].trim()) && Permission.DENY == Permission.parsePermFromString(items[1].trim())) {
+                throw new AclException(String.format("Parse resource permission error for %s.\n" +
+                    "The expected permissions are 'SUB' or 'PUB' or 'SUB|PUB' or 'PUB|SUB'.", resource));
+            }
+        }
+    }
+
     public static boolean needAdminPerm(Integer code) {
         return ADMIN_CODE.contains(code);
     }

acl/src/main/java/org/apache/rocketmq/acl/plain/PlainPermissionManager.java

@@ -128,9 +128,12 @@ public class PlainPermissionManager {
 
         if (plainAccessConfig == null) {
             log.error("Parameter value plainAccessConfig is null,Please check your parameter");
-            return false;
+            throw new AclException("Parameter value plainAccessConfig is null, Please check your parameter");
         }
 
+        Permission.checkResourcePerms(plainAccessConfig.getTopicPerms());
+        Permission.checkResourcePerms(plainAccessConfig.getGroupPerms());
+
         Map<String, Object> aclAccessConfigMap = AclUtils.getYamlDataObject(fileHome + File.separator + fileName,
             Map.class);
         if (aclAccessConfigMap == null || aclAccessConfigMap.isEmpty()) {

acl/src/test/java/org/apache/rocketmq/acl/common/PermissionTest.java

@@ -17,6 +17,7 @@
 package org.apache.rocketmq.acl.common;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -165,4 +166,27 @@ public class PermissionTest {
         aclException.setStatus("netaddress examine scope Exception netaddress");
         Assert.assertEquals(aclException.getStatus(),"netaddress examine scope Exception netaddress");
     }
+
+    @Test
+    public void checkResourcePermsNormalTest() {
+        Permission.checkResourcePerms(null);
+        Permission.checkResourcePerms(new ArrayList<>());
+        Permission.checkResourcePerms(Arrays.asList("topicA=PUB"));
+        Permission.checkResourcePerms(Arrays.asList("topicA=PUB", "topicB=SUB", "topicC=PUB|SUB"));
+    }
+
+    @Test(expected = AclException.class)
+    public void checkResourcePermsExceptionTest1() {
+        Permission.checkResourcePerms(Arrays.asList("topicA"));
+    }
+
+    @Test(expected = AclException.class)
+    public void checkResourcePermsExceptionTest2() {
+        Permission.checkResourcePerms(Arrays.asList("topicA="));
+    }
+
+    @Test(expected = AclException.class)
+    public void checkResourcePermsExceptionTest3() {
+        Permission.checkResourcePerms(Arrays.asList("topicA=DENY1"));
+    }
 }

acl/src/test/java/org/apache/rocketmq/acl/plain/PlainAccessValidatorTest.java

@@ -546,6 +546,26 @@ public class PlainAccessValidatorTest {
         Assert.assertEquals(plainAccessValidator.updateAccessConfig(plainAccessConfig), false);
     }
 
+    @Test(expected = AclException.class)
+    public void createAndUpdateAccessAclYamlConfigExceptionTest() {
+        System.setProperty("rocketmq.home.dir", "src/test/resources");
+        System.setProperty("rocketmq.acl.plain.file", "/conf/plain_acl_update_create.yml");
+
+        PlainAccessConfig plainAccessConfig = new PlainAccessConfig();
+        plainAccessConfig.setAccessKey("RocketMQ33");
+        plainAccessConfig.setSecretKey("123456789111");
+        List<String> topicPerms = new ArrayList<String>();
+        topicPerms.add("topicB=PUB");
+        plainAccessConfig.setTopicPerms(topicPerms);
+        List<String> groupPerms = new ArrayList<String>();
+        groupPerms.add("groupC=DENY1");
+        plainAccessConfig.setGroupPerms(groupPerms);
+
+        PlainAccessValidator plainAccessValidator = new PlainAccessValidator();
+        // Create element in the acl access yaml config file
+        plainAccessValidator.updateAccessConfig(plainAccessConfig);
+    }
+
     @Test
     public void updateGlobalWhiteAddrsNormalTest() {
         System.setProperty("rocketmq.home.dir", "src/test/resources");