[ISSUE #1147]Broker will report Exception if open the aclEnable and...

[ISSUE #1147]Broker will report Exception if open the aclEnable and enableDLegerCommitLog flag at the same time. (#1149)

* [issue#1078]fix User can't use mqadmin command normally if they don't copy the tool.yml file to their related fold and AclEnable flag is closed.

* [issue#1147]Broker will report Exception if open the aclEnable and enableDLegerCommitLog flag at the same time.

* [issue#1147]Broker will report Exception if open the aclEnable and enableDLegerCommitLog flag at the same time.

* [issue#1147]Add the restriction of Acl in the acl's user guide.

* [issue#1147]Adjust some codes for acl issue that users can't open the aclEnable and use Master/Slave or Dledger the same time.

* [issue#1147]Adjust and optimize some codes for acl issue that users can't open the aclEnable and use Master/Slave or Dledger the same time.

* [issue#1147]return back for original codes for acl issue that users can't open the aclEnable and use Master/Slave or Dledger the same time.

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

@@ -16,6 +16,7 @@
  */
 package org.apache.rocketmq.acl.common;
 
+import com.alibaba.fastjson.JSONObject;
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.FileNotFoundException;
@@ -23,6 +24,10 @@ import java.io.IOException;
 import java.util.Map;
 import java.util.SortedMap;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
+import org.apache.rocketmq.remoting.RPCHook;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 import org.yaml.snakeyaml.Yaml;
 
@@ -30,6 +35,8 @@ import static org.apache.rocketmq.acl.common.SessionCredentials.CHARSET;
 
 public class AclUtils {
 
+    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME);
+
     public static byte[] combineRequestContent(RemotingCommand request, SortedMap<String, String> fieldsMap) {
         try {
             StringBuilder sb = new StringBuilder("");
@@ -139,4 +146,30 @@ public class AclUtils {
         }
     }
 
+    public static RPCHook getAclRPCHook(String fileName) {
+        JSONObject yamlDataObject = null;
+        try {
+            yamlDataObject = AclUtils.getYamlDataObject(fileName,
+                JSONObject.class);
+        } catch (Exception e) {
+            log.error("convert yaml file to data object error, ",e);
+            return null;
+        }
+
+        if (yamlDataObject == null || yamlDataObject.isEmpty()) {
+            log.warn("Cannot find conf file :{}, acl isn't be enabled." ,fileName);
+            return null;
+        }
+        
+        String accessKey = yamlDataObject.getString("accessKey");
+        String secretKey = yamlDataObject.getString("secretKey");
+
+        if (StringUtils.isBlank(accessKey) || StringUtils.isBlank(secretKey)) {
+            log.warn("AccessKey or secretKey is blank, the acl is not enabled.");
+
+            return null;
+        }
+        return new AclClientRPCHook(new SessionCredentials(accessKey,secretKey));
+    }
+
 }

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

@@ -53,11 +53,13 @@ public class PlainAccessValidator implements AccessValidator {
             accessResource.setWhiteRemoteAddress(remoteAddr);
         }
 
+        accessResource.setRequestCode(request.getCode());
+
         if (request.getExtFields() == null) {
-            throw new AclException("request's extFields value is null");
+            //If request's extFields is null,then return accessResource directly(users can use whiteAddress pattern)
+            //The following logic codes depend on the request's extFields not to be null.
+            return accessResource;
         }
-        
-        accessResource.setRequestCode(request.getCode());
         accessResource.setAccessKey(request.getExtFields().get(SessionCredentials.ACCESS_KEY));
         accessResource.setSignature(request.getExtFields().get(SessionCredentials.SIGNATURE));
         accessResource.setSecretToken(request.getExtFields().get(SessionCredentials.SECURITY_TOKEN));
@@ -116,6 +118,7 @@ public class PlainAccessValidator implements AccessValidator {
         } catch (Throwable t) {
             throw new AclException(t.getMessage(), t);
         }
+
         // Content
         SortedMap<String, String> map = new TreeMap<String, String>();
         for (Map.Entry<String, String> entry : request.getExtFields().entrySet()) {

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

@@ -21,6 +21,7 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.remoting.RPCHook;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -147,7 +148,21 @@ public class AclUtilsTest {
         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 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 incompleteContRPCHook = AclUtils.getAclRPCHook("src/test/resources/conf/plain_acl_incomplete.yml");
+        Assert.assertNull(incompleteContRPCHook);
+    }
 
 
 }

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

@@ -116,7 +116,7 @@ public class PlainAccessValidatorTest {
         plainAccessValidator.validate(accessResource);
     }
 
-    @Test(expected = AclException.class)
+    @Test
     public void validateForAdminCommandWithOutAclRPCHook() {
         RemotingCommand consumerOffsetAdminRequest = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_CONSUMER_OFFSET, null);
         plainAccessValidator.parse(consumerOffsetAdminRequest, "192.168.0.1:9876");
@@ -284,4 +284,17 @@ public class PlainAccessValidatorTest {
         PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), "192.168.1.1");
         plainAccessValidator.validate(accessResource);
     }
+
+    @Test
+    public void validateGetAllTopicConfigTest() {
+        String whiteRemoteAddress = "192.168.0.1";
+        RemotingCommand remotingCommand = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_TOPIC_CONFIG, null);
+
+        ByteBuffer buf = remotingCommand.encodeHeader();
+        buf.getInt();
+        buf = ByteBuffer.allocate(buf.limit() - buf.position()).put(buf);
+        buf.position(0);
+        PlainAccessResource accessResource = (PlainAccessResource) plainAccessValidator.parse(RemotingCommand.decode(buf), whiteRemoteAddress);
+        plainAccessValidator.validate(accessResource);
+    }
 }

acl/src/test/resources/conf/plain_acl_incomplete.yml

+# 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.
+
+## suggested format
+
+- accessKey: rocketmq2
+  secretKey: 
+  whiteRemoteAddress: 192.168.1.*
+  # if it is admin, it could access all resources
+  admin: true
\ No newline at end of file

common/src/main/java/org/apache/rocketmq/common/MixAll.java

@@ -93,6 +93,7 @@ public class MixAll {
     public static final String RMQ_SYS_TRACE_TOPIC = "RMQ_SYS_TRACE_TOPIC";
     public static final String RMQ_SYS_TRANS_OP_HALF_TOPIC = "RMQ_SYS_TRANS_OP_HALF_TOPIC";
     public static final String CID_SYS_RMQ_TRANS = "CID_RMQ_SYS_TRANS";
+    public static final String ACL_CONF_TOOLS_FILE = "/conf/tools.yml";
 
     public static String getWSAddr() {
         String wsDomainName = System.getProperty("rocketmq.namesrv.domain", DEFAULT_NAMESRV_ADDR_LOOKUP);

distribution/conf/plain_acl.yml

@@ -14,6 +14,8 @@
 #  limitations under the License.
 
 globalWhiteRemoteAddresses:
+- 10.10.103.*
+- 192.168.0.*
 
 accounts:
 - accessKey: RocketMQ

distribution/conf/tools.yml

@@ -14,6 +14,6 @@
 #  limitations under the License.
 
 
-accessKey: rocketmq
+accessKey: rocketmq2
 secretKey: 12345678
     

docs/cn/acl/user_guide.md

@@ -73,10 +73,14 @@ Broker端对权限的校验逻辑主要分为以下几步：
 （2）对于某个资源，如果有显性配置权限，则采用配置的权限；如果没有显性配置权限，则采用默认的权限；
 
 ## 5. 热加载修改后权限控制定义
-RocketrMQ的权限控制存储的默认实现是基于yml配置文件。用户可以动态修改权限控制定义的属性，而不需重新启动Broker服务节点。
-
-
+RocketMQ的权限控制存储的默认实现是基于yml配置文件。用户可以动态修改权限控制定义的属性，而不需重新启动Broker服务节点。
 
+## 6. 权限控制的使用限制
+(1)如果ACL与高可用部署(Master/Slave架构)同时启用，那么需要在Broker Master节点的distribution/conf/plain_acl.yml配置文件中
+设置全局白名单信息，即为将Slave节点的ip地址设置至Master节点plain_acl.yml配置文件的全局白名单中。
 
+(2)如果ACL与高可用部署(多副本Dledger架构)同时启用，由于出现节点宕机时，Dledger Group组内会自动选主，那么就需要将Dledger Group组
+内所有Broker节点的plain_acl.yml配置文件的白名单设置所有Broker节点的ip地址。
 
+**特别注意**在[4.5.0]版本中即使使用上面所述的白名单也无法解决开启ACL的问题，解决该问题的[PR链接](https://github.com/apache/rocketmq/pull/1149)
 

tools/src/main/java/org/apache/rocketmq/tools/command/MQAdminStartup.java

@@ -21,14 +21,10 @@ import ch.qos.logback.classic.joran.JoranConfigurator;
 import ch.qos.logback.core.joran.spi.JoranException;
 import java.util.ArrayList;
 import java.util.List;
-import com.alibaba.fastjson.JSONObject;
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.Options;
 import org.apache.commons.cli.PosixParser;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.rocketmq.acl.common.AclClientRPCHook;
 import org.apache.rocketmq.acl.common.AclUtils;
-import org.apache.rocketmq.acl.common.SessionCredentials;
 import org.apache.rocketmq.common.MQVersion;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.remoting.RPCHook;
@@ -82,6 +78,9 @@ import org.slf4j.LoggerFactory;
 public class MQAdminStartup {
     protected static List<SubCommand> subCommandList = new ArrayList<SubCommand>();
 
+    private static String rocketmqHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY,
+        System.getenv(MixAll.ROCKETMQ_HOME_ENV));
+
     public static void main(String[] args) {
         main0(args, null);
     }
@@ -132,7 +131,7 @@ public class MQAdminStartup {
                             System.setProperty(MixAll.NAMESRV_ADDR_PROPERTY, namesrvAddr);
                         }
 
-                        cmd.execute(commandLine, options, getAclRPCHook());
+                        cmd.execute(commandLine, options, AclUtils.getAclRPCHook(rocketmqHome + MixAll.ACL_CONF_TOOLS_FILE));
                     } else {
                         System.out.printf("The sub command %s not exist.%n", args[0]);
                     }
@@ -203,8 +202,6 @@ public class MQAdminStartup {
     }
 
     private static void initLogback() throws JoranException {
-        String rocketmqHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY, System.getenv(MixAll.ROCKETMQ_HOME_ENV));
-
         LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();
         JoranConfigurator configurator = new JoranConfigurator();
         configurator.setContext(lc);
@@ -245,36 +242,4 @@ public class MQAdminStartup {
     public static void initCommand(SubCommand command) {
         subCommandList.add(command);
     }
-
-    public static RPCHook getAclRPCHook() {
-        String fileHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY, System.getenv(MixAll.ROCKETMQ_HOME_ENV));
-        String fileName = "/conf/tools.yml";
-        JSONObject yamlDataObject = null;
-        try {
-            yamlDataObject = AclUtils.getYamlDataObject(fileHome + fileName,
-                JSONObject.class);
-        } catch (Exception e) {
-            e.printStackTrace();
-            return null;
-        }
-
-        if (yamlDataObject == null) {
-            System.out.printf("Cannot find conf file %s, acl isn't be enabled.%n" ,fileHome + fileName);
-            return null;
-        }
-
-        if (yamlDataObject.isEmpty()) {
-            System.out.printf("Content of conf file %s is empty, acl isn't be enabled.%n" ,fileHome + fileName);
-            return null;
-        }
-
-        String accessKey = yamlDataObject.getString("accessKey");
-        String secretKey = yamlDataObject.getString("secretKey");
-
-        if (StringUtils.isBlank(accessKey) || StringUtils.isBlank(secretKey)) {
-            System.out.printf("AccessKey or secretKey is blank, the acl is not enabled.%n");
-            return null;
-        }
-        return new AclClientRPCHook(new SessionCredentials(accessKey,secretKey));
-    }
 }