Merge cf21c08 into be7c6dd

CONTRIBUTING.md

@@ -10,7 +10,7 @@ Nor is code the only way to contribute to the project. We strongly value documen
 To submit a change for inclusion, please do the following:
 
 #### If the change is non-trivial please include some unit tests that cover the new functionality.
-#### If you are introducing a completely new feature or API it is a good idea to start a wiki and get consensus on the basic design first.
+#### If you are introducing a completely new feature or API it is a good idea to start a [RIP](https://github.com/apache/rocketmq/wiki/RocketMQ-Improvement-Proposal) and get consensus on the basic design first.
 #### It is our job to follow up on patches in a timely fashion. Nag us if we aren't doing our job (sometimes we drop things).
 
 ## Becoming a Committer
@@ -19,9 +19,8 @@ We are always interested in adding new contributors. What we look for are series
 
 Nowadays,we have several important contribution points:
 #### Wiki & JavaDoc
-#### RocketMQ Console
 #### RocketMQ SDK(C++\.Net\Php\Python\Go\Node.js)
-#### RocketMQ MySQL(Oracle\PostgreSQL\Redis\MongoDB\HBase\MSSQL) Replicator
+#### RocketMQ Connectors
 
 ##### Prerequisite
 If you want to contribute the above listing points, you must abide our some prerequisites:

README.md

@@ -51,3 +51,23 @@ We always welcome new contributions, whether for trivial cleanups, [big new feat
 [Apache License, Version 2.0](http://www.apache.org/licenses/LICENSE-2.0.html) Copyright (C) Apache Software Foundation
 
 
+----------
+## Export Control Notice
+This distribution includes cryptographic software. The country in which you currently reside may have
+restrictions on the import, possession, use, and/or re-export to another country, of encryption software.
+BEFORE using any encryption software, please check your country's laws, regulations and policies concerning
+the import, possession, or use, and re-export of encryption software, to see if this is permitted. See
+<http://www.wassenaar.org/> for more information.
+
+The U.S. Government Department of Commerce, Bureau of Industry and Security (BIS), has classified this
+software as Export Commodity Control Number (ECCN) 5D002.C.1, which includes information security software
+using or performing cryptographic functions with asymmetric algorithms. The form and manner of this Apache
+Software Foundation distribution makes it eligible for export under the License Exception ENC Technology
+Software Unrestricted (TSU) exception (see the BIS Export Administration Regulations, Section 740.13) for
+both object code and source code.
+
+The following provides more details on the included cryptographic software:
+
+This software uses Apache Commons Crypto (https://commons.apache.org/proper/commons-crypto/) to
+support authentication, and encryption and decryption of data sent across the network between
+services.

acl/pom.xml

@@ -13,7 +13,7 @@
     <parent>
         <groupId>org.apache.rocketmq</groupId>
         <artifactId>rocketmq-all</artifactId>
-        <version>4.5.2-SNAPSHOT</version>
+        <version>4.6.0-SNAPSHOT</version>
     </parent>
     <artifactId>rocketmq-acl</artifactId>
     <name>rocketmq-acl ${project.version}</name>
@@ -67,6 +67,10 @@
             <artifactId>logback-core</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>commons-validator</groupId>
+            <artifactId>commons-validator</artifactId>
+        </dependency>
 
     </dependencies>
 </project>

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

@@ -23,6 +23,7 @@
 import java.io.FileWriter;
 import java.io.IOException;
 import java.io.PrintWriter;
+import java.util.ArrayList;
 import java.util.Map;
 import java.util.SortedMap;
 import org.apache.commons.lang3.StringUtils;
@@ -69,24 +70,75 @@ public static String calSignature(byte[] data, String secretKey) {
         return signature;
     }
 
+    public static void IPv6AddressCheck(String netaddress) {
+        if (isAsterisk(netaddress) || isMinus(netaddress)) {
+            int asterisk = netaddress.indexOf("*");
+            int minus = netaddress.indexOf("-");
+//            '*' must be the end of netaddress if it exists
+            if (asterisk > -1 && asterisk != netaddress.length() - 1) {
+                throw new AclException(String.format("Netaddress examine scope Exception netaddress is %s", netaddress));
+            }
+
+//            format like "2::ac5:78:1-200:*" or "2::ac5:78:1-200" is legal
+            if (minus > -1) {
+                if (asterisk == -1) {
+                    if (minus <= netaddress.lastIndexOf(":")) {
+                        throw new AclException(String.format("Netaddress examine scope Exception netaddress is %s", netaddress));
+                    }
+                } else {
+                    if (minus <= netaddress.lastIndexOf(":", netaddress.lastIndexOf(":") - 1)) {
+                        throw new AclException(String.format("Netaddress examine scope Exception netaddress is %s", netaddress));
+                    }
+                }
+            }
+        }
+    }
+
+    public static String v6ipProcess(String netaddress, String[] strArray, int index) {
+        int part;
+        String subAddress;
+        boolean isAsterisk = isAsterisk(netaddress);
+        boolean isMinus = isMinus(netaddress);
+        if (isAsterisk && isMinus) {
+            part = 6;
+            int lastColon = netaddress.lastIndexOf(':');
+            int secondLastColon = netaddress.substring(0, lastColon).lastIndexOf(':');
+            subAddress = netaddress.substring(0, secondLastColon);
+        } else if (!isAsterisk && !isMinus) {
+            part = 8;
+            subAddress = netaddress;
+        } else {
+            part = 7;
+            subAddress = netaddress.substring(0, netaddress.lastIndexOf(':'));
+        }
+        return expandIP(subAddress, part);
+    }
+
     public static void verify(String netaddress, int index) {
         if (!AclUtils.isScope(netaddress, index)) {
             throw new AclException(String.format("Netaddress examine scope Exception netaddress is %s", netaddress));
         }
     }
 
-    public static String[] getAddreeStrArray(String netaddress, String four) {
-        String[] fourStrArray = StringUtils.split(four.substring(1, four.length() - 1), ",");
+    public static String[] getAddreeStrArray(String netaddress, String partialAddress) {
+        String[] parAddStrArray = StringUtils.split(partialAddress.substring(1, partialAddress.length() - 1), ",");
         String address = netaddress.substring(0, netaddress.indexOf("{"));
-        String[] addreeStrArray = new String[fourStrArray.length];
-        for (int i = 0; i < fourStrArray.length; i++) {
-            addreeStrArray[i] = address + fourStrArray[i];
+        String[] addreeStrArray = new String[parAddStrArray.length];
+        for (int i = 0; i < parAddStrArray.length; i++) {
+            addreeStrArray[i] = address + parAddStrArray[i];
         }
         return addreeStrArray;
     }
 
-    public static boolean isScope(String num, int index) {
-        String[] strArray = StringUtils.split(num, ".");
+    public static boolean isScope(String netaddress, int index) {
+//        IPv6 Address
+        if (isColon(netaddress)) {
+            netaddress = expandIP(netaddress, 8);
+            String[] strArray = StringUtils.split(netaddress, ":");
+            return isIPv6Scope(strArray, index);
+        }
+
+        String[] strArray = StringUtils.split(netaddress, ".");
         if (strArray.length != 4) {
             return false;
         }
@@ -107,6 +159,10 @@ public static boolean isScope(String[] num, int index) {
 
     }
 
+    public static boolean isColon(String netaddress) {
+        return netaddress.indexOf(':') > -1;
+    }
+
     public static boolean isScope(String num) {
         return isScope(Integer.valueOf(num.trim()));
     }
@@ -119,7 +175,7 @@ public static boolean isAsterisk(String asterisk) {
         return asterisk.indexOf('*') > -1;
     }
 
-    public static boolean isColon(String colon) {
+    public static boolean isComma(String colon) {
         return colon.indexOf(',') > -1;
     }
 
@@ -128,6 +184,88 @@ public static boolean isMinus(String minus) {
 
     }
 
+    public static boolean isIPv6Scope(String[] num, int index) {
+        for (int i = 0; i < index; i++) {
+            int value;
+            try {
+                value = Integer.parseInt(num[i], 16);
+            } catch (NumberFormatException e) {
+                return false;
+            }
+            if (!isIPv6Scope(value)) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    public static boolean isIPv6Scope(int num) {
+        int min = Integer.parseInt("0", 16);
+        int max = Integer.parseInt("ffff", 16);
+        return num >= min && num <= max;
+    }
+
+    public static String expandIP(String netaddress, int part) {
+        boolean compress = false;
+        int compressIndex = -1;
+        String[] strArray = StringUtils.split(netaddress, ":");
+        ArrayList<Integer> indexes = new ArrayList<>();
+        for (int i = 0; i < netaddress.length(); i++) {
+            if (netaddress.charAt(i) == ':') {
+                if (indexes.size() > 0 && i - indexes.get(indexes.size() - 1) == 1) {
+                    compressIndex = i;
+                    compress = true;
+                }
+                indexes.add(i);
+            }
+        }
+
+        for (int i = 0; i < strArray.length; i++) {
+            if (strArray[i].length() < 4) {
+                strArray[i] = "0000".substring(0, 4 - strArray[i].length()) + strArray[i];
+            }
+        }
+
+        StringBuilder sb = new StringBuilder();
+        if (compress) {
+            int pos = indexes.indexOf(compressIndex);
+            int index = 0;
+            if (!netaddress.startsWith(":")) {
+                for (int i = 0; i < pos; i++) {
+                    sb.append(strArray[index]).append(":");
+                    index += 1;
+                }
+            }
+            int zeroNum = part - strArray.length;
+            if (netaddress.endsWith(":")) {
+                for (int i = 0; i < zeroNum; i++) {
+                    sb.append("0000");
+                    if (i != zeroNum - 1) {
+                        sb.append(":");
+                    }
+                }
+            } else {
+                for (int i = 0; i < zeroNum; i++) {
+                    sb.append("0000").append(":");
+                }
+                for (int i = index; i < strArray.length; i++) {
+                    sb.append(strArray[i]);
+                    if (i != strArray.length - 1) {
+                        sb.append(":");
+                    }
+                }
+            }
+        } else {
+            for (int i = 0; i < strArray.length; i++) {
+                sb.append(strArray[i]);
+                if (i != strArray.length - 1) {
+                    sb.append(":");
+                }
+            }
+        }
+        return sb.toString().toUpperCase();
+    }
+
     public static <T> T getYamlDataObject(String path, Class<T> clazz) {
         Yaml yaml = new Yaml();
         FileInputStream fis = null;
@@ -148,7 +286,7 @@ public static <T> T getYamlDataObject(String path, Class<T> clazz) {
         }
     }
 
-    public static boolean writeDataObject(String path, Map<String,Object> dataMap) {
+    public static boolean writeDataObject(String path, Map<String, Object> dataMap) {
         Yaml yaml = new Yaml();
         PrintWriter pw = null;
         try {
@@ -172,15 +310,15 @@ public static RPCHook getAclRPCHook(String fileName) {
             yamlDataObject = AclUtils.getYamlDataObject(fileName,
                 JSONObject.class);
         } catch (Exception e) {
-            log.error("Convert yaml file to data object error, ",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);
+            log.warn("Cannot find conf file :{}, acl isn't be enabled.", fileName);
             return null;
         }
-        
+
         String accessKey = yamlDataObject.getString(AclConstants.CONFIG_ACCESS_KEY);
         String secretKey = yamlDataObject.getString(AclConstants.CONFIG_SECRET_KEY);
 
@@ -189,7 +327,7 @@ public static RPCHook getAclRPCHook(String fileName) {
 
             return null;
         }
-        return new AclClientRPCHook(new SessionCredentials(accessKey,secretKey));
+        return new AclClientRPCHook(new SessionCredentials(accessKey, secretKey));
     }
 
 }

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

@@ -50,7 +50,7 @@ public PlainAccessValidator() {
     public AccessResource parse(RemotingCommand request, String remoteAddr) {
         PlainAccessResource accessResource = new PlainAccessResource();
         if (remoteAddr != null && remoteAddr.contains(":")) {
-            accessResource.setWhiteRemoteAddress(remoteAddr.split(":")[0]);
+            accessResource.setWhiteRemoteAddress(remoteAddr.substring(0, remoteAddr.lastIndexOf(':')));
         } else {
             accessResource.setWhiteRemoteAddress(remoteAddr);
         }