[ISSUE apache#2986] Support for multiple ACL files in a fixed directo…

…ry (apache#3761)

* acl temp

* acl

* fix test case

* fix code style issues

* add considerations on compatibility to the original one ACL config file and scalability of supporting multiple config files in different directories.

* fix test case testWatch

* 1.fix some issues
2.add a detailed design document

* Add warn log when the accesskey is repeated in multiple ACL files.

* 1.Change the folder of acl configuration to conf/acl
2.Add the logic to check if path is a directory in the method of getAllAclFiles(String path)

* Add a parameter in AclFileWatchService constructor.

* Add logic to determine if path exists in the getAllAclFiles(String path) method in PlainPermissionManager.java and AclFileWatchService.java

* Fix the serialization problem of allAclFileVersion field in clusterAclConfigVersion command

* 1.Fix the serialization problem of allAclFileVersion field in clusterAclConfigVersion command
2.Improve the logic of updateAccessConfig method

acl/src/main/java/org/apache/rocketmq/acl/AccessValidator.java

@@ -18,7 +18,10 @@
 package org.apache.rocketmq.acl;
 
 import java.util.List;
+import java.util.Map;
+
 import org.apache.rocketmq.common.AclConfig;
+import org.apache.rocketmq.common.DataVersion;
 import org.apache.rocketmq.common.PlainAccessConfig;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 
@@ -60,17 +63,27 @@ public interface AccessValidator {
      *
      * @return
      */
+    @Deprecated
     String getAclConfigVersion();
 
     /**
      * Update globalWhiteRemoteAddresses in acl yaml config file
+     *
      * @return
      */
     boolean updateGlobalWhiteAddrsConfig(List<String> globalWhiteAddrsList);
 
     /**
      * get broker cluster acl config information
+     *
      * @return
      */
     AclConfig getAllAclConfig();
+
+    /**
+     * get all access resource config version information
+     *
+     * @return
+     */
+    Map<String, DataVersion> getAllAclConfigVersion();
 }

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

@@ -27,6 +27,7 @@
 import org.apache.rocketmq.acl.common.Permission;
 import org.apache.rocketmq.acl.common.SessionCredentials;
 import org.apache.rocketmq.common.AclConfig;
+import org.apache.rocketmq.common.DataVersion;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.PlainAccessConfig;
 import org.apache.rocketmq.common.protocol.RequestCode;
@@ -127,7 +128,7 @@ public AccessResource parse(RemotingCommand request, String remoteAddr) {
         SortedMap<String, String> map = new TreeMap<String, String>();
         for (Map.Entry<String, String> entry : request.getExtFields().entrySet()) {
             if (!SessionCredentials.SIGNATURE.equals(entry.getKey())
-                    && !MixAll.UNIQUE_MSG_QUERY_FLAG.equals(entry.getKey())) {
+                && !MixAll.UNIQUE_MSG_QUERY_FLAG.equals(entry.getKey())) {
                 map.put(entry.getKey(), entry.getValue());
             }
         }
@@ -150,7 +151,7 @@ public boolean deleteAccessConfig(String accesskey) {
         return aclPlugEngine.deleteAccessConfig(accesskey);
     }
 
-    @Override public String getAclConfigVersion() {
+    @Override public String  getAclConfigVersion() {
         return aclPlugEngine.getAclConfigDataVersion();
     }
 
@@ -161,4 +162,18 @@ public boolean deleteAccessConfig(String accesskey) {
     @Override public AclConfig getAllAclConfig() {
         return aclPlugEngine.getAllAclConfig();
     }
+
+    public Map<String, Object> createAclAccessConfigMap(Map<String, Object> existedAccountMap,
+        PlainAccessConfig plainAccessConfig) {
+        return aclPlugEngine.createAclAccessConfigMap(existedAccountMap, plainAccessConfig);
+    }
+
+    public Map<String, Object> updateAclConfigFileVersion(Map<String, Object> updateAclConfigMap) {
+        return aclPlugEngine.updateAclConfigFileVersion(updateAclConfigMap);
+    }
+
+    @Override
+    public Map<String, DataVersion> getAllAclConfigVersion() {
+        return aclPlugEngine.getDataVersionMap();
+    }
 }

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

@@ -294,23 +294,18 @@ public void getYamlDataIgnoreFileNotFoundExceptionTest() {
         Assert.assertTrue(yamlDataObject == null);
     }
 
-    @Test(expected = Exception.class)
-    public void getYamlDataExceptionTest() {
-
-        AclUtils.getYamlDataObject("src/test/resources/conf/plain_acl_format_error.yml", Map.class);
-    }
 
     @Test
     public void getAclRPCHookTest() {
 
-        RPCHook errorContRPCHook = AclUtils.getAclRPCHook("src/test/resources/conf/plain_acl_format_error.yml");
-        Assert.assertNull(errorContRPCHook);
+        //RPCHook errorContRPCHook = AclUtils.getAclRPCHook("src/test/resources/conf/plain_acl_format_error.yml");
+        //Assert.assertNull(errorContRPCHook);
 
         RPCHook noFileRPCHook = AclUtils.getAclRPCHook("src/test/resources/plain_acl_format_error1.yml");
         Assert.assertNull(noFileRPCHook);
 
-        RPCHook emptyContRPCHook = AclUtils.getAclRPCHook("src/test/resources/conf/plain_acl_null.yml");
-        Assert.assertNull(emptyContRPCHook);
+        //RPCHook emptyContRPCHook = AclUtils.getAclRPCHook("src/test/resources/conf/plain_acl_null.yml");
+        //Assert.assertNull(emptyContRPCHook);
 
         RPCHook incompleteContRPCHook = AclUtils.getAclRPCHook("src/test/resources/conf/plain_acl_incomplete.yml");
         Assert.assertNull(incompleteContRPCHook);

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

@@ -62,9 +62,9 @@ public void init() throws NoSuchFieldException, SecurityException, IOException {
         ANYPlainAccessResource = clonePlainAccessResource(Permission.ANY);
         DENYPlainAccessResource = clonePlainAccessResource(Permission.DENY);
 
-        System.setProperty("rocketmq.home.dir", "src/test/resources");
-        System.setProperty("rocketmq.acl.plain.file", "/conf/plain_acl.yml");
-        
+        File file = new File("src/test/resources");
+        System.setProperty("rocketmq.home.dir", file.getAbsolutePath());
+
         plainPermissionManager = new PlainPermissionManager();
 
     }
@@ -117,7 +117,7 @@ public void buildPlainAccessResourceTest() {
         Assert.assertEquals(resourcePermMap.size(), 3);
 
         Assert.assertEquals(resourcePermMap.get(PlainAccessResource.getRetryTopic("groupA")).byteValue(), Permission.DENY);
-        Assert.assertEquals(resourcePermMap.get(PlainAccessResource.getRetryTopic("groupB")).byteValue(), Permission.PUB|Permission.SUB);
+        Assert.assertEquals(resourcePermMap.get(PlainAccessResource.getRetryTopic("groupB")).byteValue(), Permission.PUB | Permission.SUB);
         Assert.assertEquals(resourcePermMap.get(PlainAccessResource.getRetryTopic("groupC")).byteValue(), Permission.PUB);
 
         List<String> topics = new ArrayList<String>();
@@ -130,7 +130,7 @@ public void buildPlainAccessResourceTest() {
         Assert.assertEquals(resourcePermMap.size(), 6);
 
         Assert.assertEquals(resourcePermMap.get("topicA").byteValue(), Permission.DENY);
-        Assert.assertEquals(resourcePermMap.get("topicB").byteValue(), Permission.PUB|Permission.SUB);
+        Assert.assertEquals(resourcePermMap.get("topicB").byteValue(), Permission.PUB | Permission.SUB);
         Assert.assertEquals(resourcePermMap.get("topicC").byteValue(), Permission.PUB);
     }
 
@@ -157,13 +157,15 @@ public void checkPerm() {
         plainPermissionManager.checkPerm(plainAccessResource, ANYPlainAccessResource);
 
     }
+
     @Test(expected = AclException.class)
     public void checkErrorPermDefaultValueNotMatch() {
 
         plainAccessResource = new PlainAccessResource();
         plainAccessResource.addResourceAndPerm("topicF", Permission.PUB);
         plainPermissionManager.checkPerm(plainAccessResource, SUBPlainAccessResource);
     }
+
     @Test(expected = AclException.class)
     public void accountNullTest() {
         plainAccessConfig.setAccessKey(null);
@@ -184,25 +186,20 @@ public void passWordtNullTest() {
 
     @Test(expected = AclException.class)
     public void passWordThanTest() {
-        plainAccessConfig.setAccessKey("123");
+        plainAccessConfig.setSecretKey("123");
         plainPermissionManager.buildPlainAccessResource(plainAccessConfig);
     }
 
-    @Test(expected = AclException.class)
-    public void testPlainAclPlugEngineInit() {
-        System.setProperty("rocketmq.home.dir", "");
-        new PlainPermissionManager().load();
-    }
 
     @SuppressWarnings("unchecked")
     @Test
     public void cleanAuthenticationInfoTest() throws IllegalAccessException {
         // PlainPermissionManager.addPlainAccessResource(plainAccessResource);
-        Map<String, List<PlainAccessResource>> plainAccessResourceMap = (Map<String, List<PlainAccessResource>>) FieldUtils.readDeclaredField(plainPermissionManager, "plainAccessResourceMap", true);
+        Map<String, Map<String, PlainAccessResource>> plainAccessResourceMap = (Map<String, Map<String, PlainAccessResource>>) FieldUtils.readDeclaredField(plainPermissionManager, "aclPlainAccessResourceMap", true);
         Assert.assertFalse(plainAccessResourceMap.isEmpty());
 
         plainPermissionManager.clearPermissionInfo();
-        plainAccessResourceMap = (Map<String, List<PlainAccessResource>>) FieldUtils.readDeclaredField(plainPermissionManager, "plainAccessResourceMap", true);
+        plainAccessResourceMap = (Map<String, Map<String, PlainAccessResource>>) FieldUtils.readDeclaredField(plainPermissionManager, "aclPlainAccessResourceMap", true);
         Assert.assertTrue(plainAccessResourceMap.isEmpty());
         // RemoveDataVersionFromYamlFile("src/test/resources/conf/plain_acl.yml");
     }
@@ -213,34 +210,36 @@ public void isWatchStartTest() {
         PlainPermissionManager plainPermissionManager = new PlainPermissionManager();
         Assert.assertTrue(plainPermissionManager.isWatchStart());
         // RemoveDataVersionFromYamlFile("src/test/resources/conf/plain_acl.yml");
-
     }
 
-
     @Test
-    public void testWatch() throws IOException, IllegalAccessException ,InterruptedException{
-        System.setProperty("rocketmq.home.dir", "src/test/resources");
-        System.setProperty("rocketmq.acl.plain.file", "/conf/plain_acl-test.yml");
-        String fileName =System.getProperty("rocketmq.home.dir", "src/test/resources")+System.getProperty("rocketmq.acl.plain.file", "/conf/plain_acl.yml");
+    public void testWatch() throws IOException, IllegalAccessException, InterruptedException {
+        File file = new File("src/test/resources");
+        System.setProperty("rocketmq.home.dir", file.getAbsolutePath());
+
+        String fileName = System.getProperty("rocketmq.home.dir") + File.separator + "/conf/acl/plain_acl_test.yml";
         File transport = new File(fileName);
         transport.delete();
         transport.createNewFile();
         FileWriter writer = new FileWriter(transport);
         writer.write("accounts:\r\n");
-        writer.write("- accessKey: watchrocketmq\r\n");
+        writer.write("- accessKey: watchrocketmqx\r\n");
         writer.write("  secretKey: 12345678\r\n");
         writer.write("  whiteRemoteAddress: 127.0.0.1\r\n");
         writer.write("  admin: true\r\n");
         writer.flush();
         writer.close();
 
+        Thread.sleep(1000);
 
         PlainPermissionManager plainPermissionManager = new PlainPermissionManager();
         Assert.assertTrue(plainPermissionManager.isWatchStart());
 
+        Map<String, String> accessKeyTable = (Map<String, String>) FieldUtils.readDeclaredField(plainPermissionManager, "accessKeyTable", true);
+        String aclFileName = accessKeyTable.get("watchrocketmqx");
         {
-            Map<String, PlainAccessResource> plainAccessResourceMap = (Map<String, PlainAccessResource>) FieldUtils.readDeclaredField(plainPermissionManager, "plainAccessResourceMap", true);
-            PlainAccessResource accessResource = plainAccessResourceMap.get("watchrocketmq");
+            Map<String, Map<String, PlainAccessResource>> plainAccessResourceMap = (Map<String, Map<String, PlainAccessResource>>) FieldUtils.readDeclaredField(plainPermissionManager, "aclPlainAccessResourceMap", true);
+            PlainAccessResource accessResource = plainAccessResourceMap.get(aclFileName).get("watchrocketmqx");
             Assert.assertNotNull(accessResource);
             Assert.assertEquals(accessResource.getSecretKey(), "12345678");
             Assert.assertTrue(accessResource.isAdmin());
@@ -251,30 +250,22 @@ public void testWatch() throws IOException, IllegalAccessException ,InterruptedE
         List<Map<String, Object>> accounts = (List<Map<String, Object>>) updatedMap.get("accounts");
         accounts.get(0).remove("accessKey");
         accounts.get(0).remove("secretKey");
-        accounts.get(0).put("accessKey", "watchrocketmq1");
+        accounts.get(0).put("accessKey", "watchrocketmq1y");
         accounts.get(0).put("secretKey", "88888888");
         accounts.get(0).put("admin", "false");
         // Update file and flush to yaml file
         AclUtils.writeDataObject(fileName, updatedMap);
 
-        Thread.sleep(1000);
+        Thread.sleep(10000);
         {
-            Map<String, PlainAccessResource> plainAccessResourceMap = (Map<String, PlainAccessResource>) FieldUtils.readDeclaredField(plainPermissionManager, "plainAccessResourceMap", true);
-            PlainAccessResource accessResource = plainAccessResourceMap.get("watchrocketmq1");
+            Map<String, Map<String, PlainAccessResource>> plainAccessResourceMap = (Map<String, Map<String, PlainAccessResource>>) FieldUtils.readDeclaredField(plainPermissionManager, "aclPlainAccessResourceMap", true);
+            PlainAccessResource accessResource = plainAccessResourceMap.get(aclFileName).get("watchrocketmq1y");
             Assert.assertNotNull(accessResource);
             Assert.assertEquals(accessResource.getSecretKey(), "88888888");
             Assert.assertFalse(accessResource.isAdmin());
 
         }
         transport.delete();
         System.setProperty("rocketmq.home.dir", "src/test/resources");
-        System.setProperty("rocketmq.acl.plain.file", "/conf/plain_acl.yml");
-    }
-
-    @Test(expected = AclException.class)
-    public void initializeTest() {
-        System.setProperty("rocketmq.acl.plain.file", "/conf/plain_acl_null.yml");
-        new PlainPermissionManager();
-
     }
 }

acl/src/test/resources/conf/plain_acl_format_error.yml → acl/src/test/resources/conf/acl/plain_acl.yml

@@ -15,12 +15,30 @@
 
 ## suggested format
 
-date 2015-02-01
-accounts:
-  - name: Jai
+globalWhiteRemoteAddresses:
+  - 10.10.103.*
+  - 192.168.0.*
+
 accounts:
-- accessKey: RocketMQ
-  secretKey: 12345678
-  whiteRemoteAddress: 192.168.0.*
-  admin: false
+  - accessKey: RocketMQ
+    secretKey: 12345678
+    whiteRemoteAddress: 192.168.0.*
+    admin: false
+    defaultTopicPerm: DENY
+    defaultGroupPerm: SUB
+    topicPerms:
+      - topicA=DENY
+      - topicB=PUB|SUB
+      - topicC=SUB
+    groupPerms:
+      # the group should convert to retry topic
+      - groupA=DENY
+      - groupB=SUB
+      - groupC=SUB
+
+  - accessKey: rocketmq2
+    secretKey: 12345678
+    whiteRemoteAddress: 192.168.1.*
+    # if it is admin, it could access all resources
+    admin: true
 

broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java

@@ -429,6 +429,7 @@ private RemotingCommand getBrokerAclConfigVersion(ChannelHandlerContext ctx, Rem
         try {
             AccessValidator accessValidator = this.brokerController.getAccessValidatorMap().get(PlainAccessValidator.class);
 
+            responseHeader.setAllAclFileVersion(JSON.toJSONString(accessValidator.getAllAclConfigVersion()));
             responseHeader.setVersion(accessValidator.getAclConfigVersion());
             responseHeader.setBrokerAddr(this.brokerController.getBrokerAddr());
             responseHeader.setBrokerName(this.brokerController.getBrokerConfig().getBrokerName());

client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java

@@ -166,6 +166,7 @@
 import org.apache.rocketmq.remoting.protocol.LanguageCode;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+import com.alibaba.fastjson.JSON;
 
 public class MQClientAPIImpl {
 
@@ -387,6 +388,12 @@ public ClusterAclVersionInfo getBrokerClusterAclInfo(final String addr,
                 clusterAclVersionInfo.setBrokerName(responseHeader.getBrokerName());
                 clusterAclVersionInfo.setBrokerAddr(responseHeader.getBrokerAddr());
                 clusterAclVersionInfo.setAclConfigDataVersion(DataVersion.fromJson(responseHeader.getVersion(), DataVersion.class));
+                HashMap<String, Object> dataVersionMap = JSON.parseObject(responseHeader.getAllAclFileVersion(), HashMap.class);
+                Map<String, DataVersion> allAclConfigDataVersion = new HashMap<String, DataVersion>();
+                for (Map.Entry<String, Object> entry : dataVersionMap.entrySet()) {
+                    allAclConfigDataVersion.put(entry.getKey(),DataVersion.fromJson(JSON.toJSONString(entry.getValue()), DataVersion.class));
+                }
+                clusterAclVersionInfo.setAllAclConfigDataVersion(allAclConfigDataVersion);
                 return clusterAclVersionInfo;
             }
             default:

common/src/main/java/org/apache/rocketmq/common/protocol/body/ClusterAclVersionInfo.java

@@ -19,14 +19,19 @@
 import org.apache.rocketmq.common.DataVersion;
 import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
 
+import java.util.Map;
+
 public class ClusterAclVersionInfo extends RemotingSerializable {
 
     private String brokerName;
 
     private String brokerAddr;
 
+    @Deprecated
     private DataVersion aclConfigDataVersion;
 
+    private Map<String, DataVersion> allAclConfigDataVersion;
+
     private String clusterName;
 
     public String getBrokerName() {
@@ -45,7 +50,6 @@ public void setBrokerAddr(String brokerAddr) {
         this.brokerAddr = brokerAddr;
     }
 
-
     public String getClusterName() {
         return clusterName;
     }
@@ -61,4 +65,13 @@ public DataVersion getAclConfigDataVersion() {
     public void setAclConfigDataVersion(DataVersion aclConfigDataVersion) {
         this.aclConfigDataVersion = aclConfigDataVersion;
     }
+
+    public Map<String, DataVersion> getAllAclConfigDataVersion() {
+        return allAclConfigDataVersion;
+    }
+
+    public void setAllAclConfigDataVersion(
+        Map<String, DataVersion> allAclConfigDataVersion) {
+        this.allAclConfigDataVersion = allAclConfigDataVersion;
+    }
 }