Add signature and redesign the permission graph

.gitignore

@@ -10,4 +10,6 @@ devenv
 *.versionsBackup
 !NOTICE-BIN
 !LICENSE-BIN
-.DS_Store
\ No newline at end of file
+.DS_Store
+localbin
+nohup.out

acl/pom.xml

@@ -30,7 +30,6 @@
             <groupId>${project.groupId}</groupId>
             <artifactId>rocketmq-remoting</artifactId>
         </dependency>
-
         <dependency>
             <groupId>${project.groupId}</groupId>
             <artifactId>rocketmq-logging</artifactId>
@@ -43,6 +42,10 @@
             <groupId>org.yaml</groupId>
             <artifactId>snakeyaml</artifactId>
         </dependency>
+        <dependency>
+            <groupId>commons-codec</groupId>
+            <artifactId>commons-codec</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-lang3</artifactId>

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

-/*
- * 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.
- */
-package org.apache.rocketmq.acl;
-
-import java.util.HashMap;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.rocketmq.acl.plug.AccessControl;
-import org.apache.rocketmq.acl.plug.AclPlugRuntimeException;
-import org.apache.rocketmq.acl.plug.AuthenticationResult;
-import org.apache.rocketmq.acl.plug.PlainAclPlugEngine;
-import org.apache.rocketmq.remoting.protocol.RemotingCommand;
-
-public class PlainAccessValidator implements AccessValidator {
-
-    private PlainAclPlugEngine aclPlugEngine;
-
-    public PlainAccessValidator() {
-        aclPlugEngine = new PlainAclPlugEngine();
-    }
-
-    @Override
-    public AccessResource parse(RemotingCommand request, String remoteAddr) {
-        HashMap<String, String> extFields = request.getExtFields();
-        int code = request.getCode();
-        AccessControl accessControl = new AccessControl();
-        accessControl.setCode(request.getCode());
-        accessControl.setRecognition(remoteAddr);
-        accessControl.setNetaddress(StringUtils.split(remoteAddr, ":")[0]);
-        if (extFields != null) {
-            accessControl.setAccount(extFields.get("account"));
-            accessControl.setPassword(extFields.get("password"));
-            if (code == 310 || code == 320) {
-                accessControl.setTopic(extFields.get("b"));
-            } else {
-                accessControl.setTopic(extFields.get("topic"));
-
-            }
-        }
-        return accessControl;
-    }
-
-    @Override
-    public void validate(AccessResource accessResource) {
-        AuthenticationResult authenticationResult = null;
-        try {
-            authenticationResult = aclPlugEngine.eachCheckAuthentication((AccessControl) accessResource);
-            if (authenticationResult.isSucceed())
-                return;
-        } catch (Exception e) {
-            throw new AclPlugRuntimeException(String.format("validate exception AccessResource data %s", accessResource.toString()), e);
-        }
-        if (authenticationResult.getException() != null) {
-            throw new AclPlugRuntimeException(String.format("eachCheck the inspection appear exception, accessControl data is %s", accessResource.toString()), authenticationResult.getException());
-        }
-        if (authenticationResult.getAccessControl() != null || !authenticationResult.isSucceed()) {
-            throw new AclPlugRuntimeException(String.format("%s accessControl data is %s", authenticationResult.getResultString(), accessResource.toString()));
-        }
-    }
-
-}

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

+package org.apache.rocketmq.acl.common;
+
+import org.apache.rocketmq.remoting.CommandCustomHeader;
+import org.apache.rocketmq.remoting.RPCHook;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import java.lang.reflect.Field;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static org.apache.rocketmq.acl.common.SessionCredentials.AccessKey;
+import static org.apache.rocketmq.acl.common.SessionCredentials.SecurityToken;
+import static org.apache.rocketmq.acl.common.SessionCredentials.Signature;
+
+public class AclClientRPCHook implements RPCHook {
+    protected ConcurrentHashMap<Class<? extends CommandCustomHeader>, Field[]> fieldCache =
+            new ConcurrentHashMap<Class<? extends CommandCustomHeader>, Field[]>();
+
+
+
+    private final SessionCredentials sessionCredentials;
+
+    public AclClientRPCHook(SessionCredentials sessionCredentials) {
+        this.sessionCredentials = sessionCredentials;
+    }
+
+    @Override
+    public void doBeforeRequest(String remoteAddr, RemotingCommand request) {
+        byte[] total = AclUtils.combineRequestContent(request,
+            parseRequestContent(request, sessionCredentials.getAccessKey(), sessionCredentials.getSecurityToken()));
+        String signature = AclUtils.calSignature(total, sessionCredentials.getSecretKey());
+        request.addExtField(Signature, signature);
+        request.addExtField(AccessKey, sessionCredentials.getAccessKey());
+
+        if (sessionCredentials.getSecurityToken() != null) {
+            request.addExtField(SecurityToken, sessionCredentials.getSecurityToken());
+        }
+    }
+
+
+    @Override
+    public void doAfterResponse(String remoteAddr, RemotingCommand request, RemotingCommand response) {
+
+    }
+
+    protected SortedMap<String, String> parseRequestContent(RemotingCommand request, String ak, String securityToken) {
+        CommandCustomHeader header = request.readCustomHeader();
+        // sort property
+        SortedMap<String, String> map = new TreeMap<String, String>();
+        map.put(AccessKey, ak);
+        if (securityToken != null) {
+            map.put(SecurityToken, securityToken);
+        }
+        try {
+            // add header properties
+            if (null != header) {
+                Field[] fields = fieldCache.get(header.getClass());
+                if (null == fields) {
+                    fields = header.getClass().getDeclaredFields();
+                    for (Field field : fields) {
+                        field.setAccessible(true);
+                    }
+                    Field[] tmp = fieldCache.putIfAbsent(header.getClass(), fields);
+                    if (null != tmp) {
+                        fields = tmp;
+                    }
+                }
+
+                for (Field field : fields) {
+                    Object value = field.get(header);
+                    if (null != value && !field.isSynthetic()) {
+                        map.put(field.getName(), value.toString());
+                    }
+                }
+            }
+            return map;
+        } catch (Exception e) {
+            throw new RuntimeException("incompatible exception.", e);
+        }
+    }
+
+    public SessionCredentials getSessionCredentials() {
+        return sessionCredentials;
+    }
+}

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

+package org.apache.rocketmq.acl.common;
+
+public class AclException extends RuntimeException {
+    private static final long serialVersionUID = 1L;
+
+    private String status;
+    private int code;
+
+
+    public AclException(String status, int code) {
+        super();
+        this.status = status;
+        this.code = code;
+    }
+
+
+    public AclException(String status, int code, String message) {
+        super(message);
+        this.status = status;
+        this.code = code;
+    }
+
+
+    public AclException(String status, int code, Throwable throwable) {
+        super(throwable);
+        this.status = status;
+        this.code = code;
+    }
+
+
+    public AclException(String status, int code, String message, Throwable throwable) {
+        super(message, throwable);
+        this.status = status;
+        this.code = code;
+    }
+
+    public String getStatus() {
+        return status;
+    }
+
+    public void setStatus(String status) {
+        this.status = status;
+    }
+
+    public int getCode() {
+        return code;
+    }
+
+    public void setCode(int code) {
+        this.code = code;
+    }
+}

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

+package org.apache.rocketmq.acl.common;
+
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.logging.InternalLogger;
+import org.apache.rocketmq.logging.InternalLoggerFactory;
+import java.nio.charset.Charset;
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+import org.apache.commons.codec.binary.Base64;
+
+
+public class AclSigner {
+    public static final Charset defaultCharset = Charset.forName("UTF-8");
+    public static final SigningAlgorithm defaultAlgorithm = SigningAlgorithm.HmacSHA1;
+    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.ROCKETMQ_AUTHORIZE_LOGGER_NAME);
+    private static final int CAL_SIGNATURE_FAILED = 10015;
+    private static final String CAL_SIGNATURE_FAILED_MSG = "[%s:signature-failed] unable to calculate a request signature. error=%s";
+
+    public static String calSignature(String data, String key) throws AclException {
+        return calSignature(data, key, defaultAlgorithm, defaultCharset);
+    }
+
+    public static String calSignature(String data, String key, SigningAlgorithm algorithm, Charset charset) throws AclException {
+        return signAndBase64Encode(data, key, algorithm, charset);
+    }
+
+    private static String signAndBase64Encode(String data, String key, SigningAlgorithm algorithm, Charset charset)
+            throws AclException {
+        try {
+            byte[] signature = sign(data.getBytes(charset), key.getBytes(charset), algorithm);
+            return new String(Base64.encodeBase64(signature), defaultCharset);
+        } catch (Exception e) {
+            String message = String.format(CAL_SIGNATURE_FAILED_MSG, CAL_SIGNATURE_FAILED, e.getMessage());
+            log.error(message, e);
+            throw new AclException("CAL_SIGNATURE_FAILED", CAL_SIGNATURE_FAILED, message, e);
+        }
+    }
+
+    private static byte[] sign(byte[] data, byte[] key, SigningAlgorithm algorithm) throws AclException {
+        try {
+            Mac mac = Mac.getInstance(algorithm.toString());
+            mac.init(new SecretKeySpec(key, algorithm.toString()));
+            return mac.doFinal(data);
+        } catch (Exception e) {
+            String message = String.format(CAL_SIGNATURE_FAILED_MSG, CAL_SIGNATURE_FAILED, e.getMessage());
+            log.error(message, e);
+            throw new AclException("CAL_SIGNATURE_FAILED", CAL_SIGNATURE_FAILED, message, e);
+        }
+    }
+
+    public static String calSignature(byte[] data, String key) throws AclException {
+        return calSignature(data, key, defaultAlgorithm, defaultCharset);
+    }
+
+    public static String calSignature(byte[] data, String key, SigningAlgorithm algorithm, Charset charset) throws AclException {
+        return signAndBase64Encode(data, key, algorithm, charset);
+    }
+
+    private static String signAndBase64Encode(byte[] data, String key, SigningAlgorithm algorithm, Charset charset)
+            throws AclException {
+        try {
+            byte[] signature = sign(data, key.getBytes(charset), algorithm);
+            return new String(Base64.encodeBase64(signature), defaultCharset);
+        } catch (Exception e) {
+            String message = String.format(CAL_SIGNATURE_FAILED_MSG, CAL_SIGNATURE_FAILED, e.getMessage());
+            log.error(message, e);
+            throw new AclException("CAL_SIGNATURE_FAILED", CAL_SIGNATURE_FAILED, message, e);
+        }
+    }
+
+}

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

@@ -14,16 +14,54 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.rocketmq.acl.plug;
+package org.apache.rocketmq.acl.common;
 
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.IOException;
+import java.util.Map;
+import java.util.SortedMap;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.acl.plain.AclPlugRuntimeException;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 import org.yaml.snakeyaml.Yaml;
 
+import static org.apache.rocketmq.acl.common.SessionCredentials.CHARSET;
+
 public class AclUtils {
 
+    public static byte[] combineRequestContent(RemotingCommand request, SortedMap<String, String> fieldsMap) {
+        try {
+            StringBuilder sb = new StringBuilder("");
+            for (Map.Entry<String, String> entry : fieldsMap.entrySet()) {
+                if (!SessionCredentials.Signature.equals(entry.getKey())) {
+                    sb.append(entry.getValue());
+                }
+            }
+
+            return AclUtils.combineBytes(sb.toString().getBytes(CHARSET), request.getBody());
+        } catch (Exception e) {
+            throw new RuntimeException("incompatible exception.", e);
+        }
+    }
+
+
+    public static byte[] combineBytes(byte[] b1, byte[] b2) {
+        int size = (null != b1 ? b1.length : 0) + (null != b2 ? b2.length : 0);
+        byte[] total = new byte[size];
+        if (null != b1)
+            System.arraycopy(b1, 0, total, 0, b1.length);
+        if (null != b2)
+            System.arraycopy(b2, 0, total, b1.length, b2.length);
+        return total;
+    }
+
+
+    public static String calSignature(byte[] data, String secretKey) {
+        String signature = AclSigner.calSignature(data, secretKey);
+        return signature;
+    }
+
     public static void verify(String netaddress, int index) {
         if (!AclUtils.isScope(netaddress, index)) {
             throw new AclPlugRuntimeException(String.format("netaddress examine scope Exception netaddress is %s", netaddress));

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

+package org.apache.rocketmq.acl.common;
+
+public class Permission {
+
+    public static final byte DENY = 1;
+    public static final byte ANY = 1 << 1;
+    public static final byte PUB = 1 << 2;
+    public static final byte SUB = 1 << 3;
+
+    public boolean checkPermission(byte neededPerm, byte ownedPerm) {
+        if ((ownedPerm & DENY) > 0) {
+            return false;
+        }
+        if ((neededPerm & ANY) > 0) {
+            return ((ownedPerm & PUB) > 0) || ((ownedPerm & SUB) > 0);
+        }
+        return (neededPerm & ownedPerm) > 0;
+    }
+
+}

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

+package org.apache.rocketmq.acl.common;
+
+import org.apache.rocketmq.common.MixAll;
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.util.Properties;
+
+public class SessionCredentials {
+    public static final Charset CHARSET = Charset.forName("UTF-8");
+    public static final String AccessKey = "AccessKey";
+    public static final String SecretKey = "SecretKey";
+    public static final String Signature = "Signature";
+    public static final String SecurityToken = "SecurityToken";
+
+    public static final String KeyFile = System.getProperty("rocketmq.client.keyFile",
+        System.getProperty("user.home") + File.separator + "onskey");
+
+    private String accessKey;
+    private String secretKey;
+    private String securityToken;
+    private String signature;
+
+    public SessionCredentials() {
+        String keyContent = null;
+        try {
+            keyContent = MixAll.file2String(KeyFile);
+        } catch (IOException ignore) {
+        }
+        if (keyContent != null) {
+            Properties prop = MixAll.string2Properties(keyContent);
+            if (prop != null) {
+                this.updateContent(prop);
+            }
+        }
+    }
+
+    public SessionCredentials(String accessKey, String secretKey) {
+        this.accessKey = accessKey;
+        this.secretKey = secretKey;
+    }
+
+    public SessionCredentials(String accessKey, String secretKey, String securityToken) {
+        this(accessKey, secretKey);
+        this.securityToken = securityToken;
+    }
+
+
+    public void updateContent(Properties prop) {
+        {
+            String value = prop.getProperty(AccessKey);
+            if (value != null) {
+                this.accessKey = value.trim();
+            }
+        }
+        {
+            String value = prop.getProperty(SecretKey);
+            if (value != null) {
+                this.secretKey = value.trim();
+            }
+        }
+        {
+            String value = prop.getProperty(SecurityToken);
+            if (value != null) {
+                this.securityToken = value.trim();
+            }
+        }
+    }
+
+    public String getAccessKey() {
+        return accessKey;
+    }
+
+    public void setAccessKey(String accessKey) {
+        this.accessKey = accessKey;
+    }
+
+    public String getSecretKey() {
+        return secretKey;
+    }
+
+    public void setSecretKey(String secretKey) {
+        this.secretKey = secretKey;
+    }
+
+    public String getSignature() {
+        return signature;
+    }
+
+    public void setSignature(String signature) {
+        this.signature = signature;
+    }
+
+    public String getSecurityToken() {
+        return securityToken;
+    }
+
+    public void setSecurityToken(final String securityToken) {
+        this.securityToken = securityToken;
+    }
+
+
+
+    @Override
+    public int hashCode() {
+        final int prime = 31;
+        int result = 1;
+        result = prime * result + ((accessKey == null) ? 0 : accessKey.hashCode());
+        result = prime * result + ((secretKey == null) ? 0 : secretKey.hashCode());
+        result = prime * result + ((signature == null) ? 0 : signature.hashCode());
+        return result;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj)
+            return true;
+        if (obj == null)
+            return false;
+        if (getClass() != obj.getClass())
+            return false;
+
+        SessionCredentials other = (SessionCredentials) obj;
+        if (accessKey == null) {
+            if (other.accessKey != null)
+                return false;
+        } else if (!accessKey.equals(other.accessKey))
+            return false;
+
+        if (secretKey == null) {
+            if (other.secretKey != null)
+                return false;
+        } else if (!secretKey.equals(other.secretKey))
+            return false;
+
+        if (signature == null) {
+            if (other.signature != null)
+                return false;
+        } else if (!signature.equals(other.signature))
+            return false;
+
+        return true;
+    }
+
+    @Override
+    public String toString() {
+        return "SessionCredentials [accessKey=" + accessKey + ", secretKey=" + secretKey + ", signature="
+            + signature + ", SecurityToken=" + securityToken + "]";
+    }
+}
\ No newline at end of file

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

+package org.apache.rocketmq.acl.common;//package com.aliyun.openservices.ons.api.impl.rocketmq.spas;
+
+public enum SigningAlgorithm {
+    HmacSHA1,
+    HmacSHA256,
+    HmacMD5;
+
+}

acl/src/main/java/org/apache/rocketmq/acl/plug/AclPlugRuntimeException.java → acl/src/main/java/org/apache/rocketmq/acl/plain/AclPlugRuntimeException.java

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.rocketmq.acl.plug;
+package org.apache.rocketmq.acl.plain;
 
 public class AclPlugRuntimeException extends RuntimeException {
 

acl/src/main/java/org/apache/rocketmq/acl/plug/AuthenticationInfo.java → acl/src/main/java/org/apache/rocketmq/acl/plain/AuthenticationInfo.java

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.rocketmq.acl.plug;
+package org.apache.rocketmq.acl.plain;
 
 import java.util.Iterator;
 import java.util.Map;
@@ -22,26 +22,26 @@ import java.util.Map.Entry;
 
 public class AuthenticationInfo {
 
-    private AccessControl accessControl;
+    private PlainAccessResource plainAccessResource;
 
     private NetaddressStrategy netaddressStrategy;
 
     private Map<Integer, Boolean> authority;
 
-    public AuthenticationInfo(Map<Integer, Boolean> authority, AccessControl accessControl,
+    public AuthenticationInfo(Map<Integer, Boolean> authority, PlainAccessResource plainAccessResource,
         NetaddressStrategy netaddressStrategy) {
         super();
         this.authority = authority;
-        this.accessControl = accessControl;
+        this.plainAccessResource = plainAccessResource;
         this.netaddressStrategy = netaddressStrategy;
     }
 
-    public AccessControl getAccessControl() {
-        return accessControl;
+    public PlainAccessResource getPlainAccessResource() {
+        return plainAccessResource;
     }
 
-    public void setAccessControl(AccessControl accessControl) {
-        this.accessControl = accessControl;
+    public void setPlainAccessResource(PlainAccessResource plainAccessResource) {
+        this.plainAccessResource = plainAccessResource;
     }
 
     public NetaddressStrategy getNetaddressStrategy() {
@@ -63,7 +63,7 @@ public class AuthenticationInfo {
     @Override
     public String toString() {
         StringBuilder builder = new StringBuilder();
-        builder.append("AuthenticationInfo [accessControl=").append(accessControl).append(", netaddressStrategy=")
+        builder.append("AuthenticationInfo [plainAccessResource=").append(plainAccessResource).append(", netaddressStrategy=")
             .append(netaddressStrategy).append(", authority={");
         Iterator<Entry<Integer, Boolean>> it = authority.entrySet().iterator();
         while (it.hasNext()) {

acl/src/main/java/org/apache/rocketmq/acl/plug/AuthenticationResult.java → acl/src/main/java/org/apache/rocketmq/acl/plain/AuthenticationResult.java

@@ -14,11 +14,11 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.rocketmq.acl.plug;
+package org.apache.rocketmq.acl.plain;
 
 public class AuthenticationResult {
 
-    private AccessControl accessControl;
+    private PlainAccessResource plainAccessResource;
 
     private boolean succeed;
 
@@ -26,12 +26,12 @@ public class AuthenticationResult {
 
     private String resultString;
 
-    public AccessControl getAccessControl() {
-        return accessControl;
+    public PlainAccessResource getPlainAccessResource() {
+        return plainAccessResource;
     }
 
-    public void setAccessControl(AccessControl accessControl) {
-        this.accessControl = accessControl;
+    public void setPlainAccessResource(PlainAccessResource plainAccessResource) {
+        this.plainAccessResource = plainAccessResource;
     }
 
     public boolean isSucceed() {

acl/src/main/java/org/apache/rocketmq/acl/plug/BrokerAccessControl.java → acl/src/main/java/org/apache/rocketmq/acl/plain/BrokerAccessControl.java

@@ -14,12 +14,12 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.rocketmq.acl.plug;
+package org.apache.rocketmq.acl.plain;
 
 import java.util.HashSet;
 import java.util.Set;
 
-public class BrokerAccessControl extends AccessControl {
+public class BrokerAccessControl extends PlainAccessResource {
 
     private boolean admin;
 

acl/src/main/java/org/apache/rocketmq/acl/plug/NetaddressStrategy.java → acl/src/main/java/org/apache/rocketmq/acl/plain/NetaddressStrategy.java

@@ -14,9 +14,9 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.rocketmq.acl.plug;
+package org.apache.rocketmq.acl.plain;
 
 public interface NetaddressStrategy {
 
-    public boolean match(AccessControl accessControl);
+    public boolean match(PlainAccessResource plainAccessResource);
 }

acl/src/main/java/org/apache/rocketmq/acl/plug/NetaddressStrategyFactory.java → acl/src/main/java/org/apache/rocketmq/acl/plain/NetaddressStrategyFactory.java

@@ -14,18 +14,19 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.rocketmq.acl.plug;
+package org.apache.rocketmq.acl.plain;
 
 import java.util.HashSet;
 import java.util.Set;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.acl.common.AclUtils;
 
 public class NetaddressStrategyFactory {
 
     public static final NullNetaddressStrategy NULL_NET_ADDRESS_STRATEGY = new NullNetaddressStrategy();
 
-    public NetaddressStrategy getNetaddressStrategy(AccessControl accessControl) {
-        String netaddress = accessControl.getNetaddress();
+    public NetaddressStrategy getNetaddressStrategy(PlainAccessResource plainAccessResource) {
+        String netaddress = plainAccessResource.getRemoteAddr();
         if (StringUtils.isBlank(netaddress) || "*".equals(netaddress)) {
             return NULL_NET_ADDRESS_STRATEGY;
         }
@@ -47,7 +48,7 @@ public class NetaddressStrategyFactory {
 
     public static class NullNetaddressStrategy implements NetaddressStrategy {
         @Override
-        public boolean match(AccessControl accessControl) {
+        public boolean match(PlainAccessResource plainAccessResource) {
             return true;
         }
 
@@ -65,8 +66,8 @@ public class NetaddressStrategyFactory {
         }
 
         @Override
-        public boolean match(AccessControl accessControl) {
-            return multipleSet.contains(accessControl.getNetaddress());
+        public boolean match(PlainAccessResource plainAccessResource) {
+            return multipleSet.contains(plainAccessResource.getRemoteAddr());
         }
 
     }
@@ -81,8 +82,8 @@ public class NetaddressStrategyFactory {
         }
 
         @Override
-        public boolean match(AccessControl accessControl) {
-            return netaddress.equals(accessControl.getNetaddress());
+        public boolean match(PlainAccessResource plainAccessResource) {
+            return netaddress.equals(plainAccessResource.getRemoteAddr());
         }
 
     }
@@ -135,8 +136,8 @@ public class NetaddressStrategyFactory {
         }
 
         @Override
-        public boolean match(AccessControl accessControl) {
-            String netAddress = accessControl.getNetaddress();
+        public boolean match(PlainAccessResource plainAccessResource) {
+            String netAddress = plainAccessResource.getRemoteAddr();
             if (netAddress.startsWith(this.head)) {
                 String value;
                 if (index == 3) {

acl/src/main/java/org/apache/rocketmq/acl/plug/AccessControl.java → acl/src/main/java/org/apache/rocketmq/acl/plain/PlainAccessResource.java

@@ -14,49 +14,68 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.rocketmq.acl.plug;
+package org.apache.rocketmq.acl.plain;
 
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.commons.lang3.builder.ToStringBuilder;
 import org.apache.rocketmq.acl.AccessResource;
+import org.apache.rocketmq.common.MixAll;
 
-public class AccessControl implements AccessResource {
+public class PlainAccessResource implements AccessResource {
+    //identify the user
+    private String accessKey;
 
-    private String account;
+    private String signature;
+    //the content to calculate the content
+    private byte[] content;
 
-    private String password;
+    private String secretToken;
 
-    private String netaddress;
+    private Map<String, Byte> resourcePermMap = new HashMap<>();
+
+    private String remoteAddr;
 
     private String recognition;
 
-    private int code;
+    private int requestCode;
+
 
+    @Deprecated
     private String topic;
 
-    public AccessControl() {
+    public PlainAccessResource() {
+    }
+
+    public void addResourceAndPerm(String resource, byte perm) {
+        if (resource == null) {
+            return;
+        }
+        resourcePermMap.put(resource, perm);
     }
 
-    public String getAccount() {
-        return account;
+    public String getAccessKey() {
+        return accessKey;
     }
 
-    public void setAccount(String account) {
-        this.account = account;
+    public void setAccessKey(String accessKey) {
+        this.accessKey = accessKey;
     }
 
-    public String getPassword() {
-        return password;
+    public String getSignature() {
+        return signature;
     }
 
-    public void setPassword(String password) {
-        this.password = password;
+    public void setSignature(String signature) {
+        this.signature = signature;
     }
 
-    public String getNetaddress() {
-        return netaddress;
+    public String getRemoteAddr() {
+        return remoteAddr;
     }
 
-    public void setNetaddress(String netaddress) {
-        this.netaddress = netaddress;
+    public void setRemoteAddr(String remoteAddr) {
+        this.remoteAddr = remoteAddr;
     }
 
     public String getRecognition() {
@@ -67,12 +86,12 @@ public class AccessControl implements AccessResource {
         this.recognition = recognition;
     }
 
-    public int getCode() {
-        return code;
+    public int getRequestCode() {
+        return requestCode;
     }
 
-    public void setCode(int code) {
-        this.code = code;
+    public void setRequestCode(int requestCode) {
+        this.requestCode = requestCode;
     }
 
     public String getTopic() {
@@ -83,13 +102,36 @@ public class AccessControl implements AccessResource {
         this.topic = topic;
     }
 
+    public String getSecretToken() {
+        return secretToken;
+    }
+
+    public void setSecretToken(String secretToken) {
+        this.secretToken = secretToken;
+    }
+
     @Override
     public String toString() {
-        StringBuilder builder = new StringBuilder();
-        builder.append("AccessControl [account=").append(account).append(", password=").append(password)
-            .append(", netaddress=").append(netaddress).append(", recognition=").append(recognition)
-            .append(", code=").append(code).append(", topic=").append(topic).append("]");
-        return builder.toString();
+        return ToStringBuilder.reflectionToString(this);
     }
 
+
+    public static boolean isRetryTopic(String topic) {
+        return (null != topic && topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX));
+    }
+
+    public static String getRetryTopic(String group) {
+        if (group == null) {
+            return null;
+        }
+        return MixAll.getRetryTopic(group);
+    }
+
+    public byte[] getContent() {
+        return content;
+    }
+
+    public void setContent(byte[] content) {
+        this.content = content;
+    }
 }

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

+/*
+ * 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.
+ */
+package org.apache.rocketmq.acl.plain;
+
+import java.util.Map;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import org.apache.rocketmq.acl.AccessResource;
+import org.apache.rocketmq.acl.AccessValidator;
+import org.apache.rocketmq.acl.common.AclUtils;
+import org.apache.rocketmq.acl.common.AclException;
+import org.apache.rocketmq.acl.common.Permission;
+import org.apache.rocketmq.acl.common.SessionCredentials;
+import org.apache.rocketmq.common.protocol.RequestCode;
+import org.apache.rocketmq.common.protocol.header.GetConsumerListByGroupRequestHeader;
+import org.apache.rocketmq.common.protocol.header.UnregisterClientRequestHeader;
+import org.apache.rocketmq.common.protocol.header.UpdateConsumerOffsetRequestHeader;
+import org.apache.rocketmq.common.protocol.heartbeat.ConsumerData;
+import org.apache.rocketmq.common.protocol.heartbeat.HeartbeatData;
+import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+
+import static org.apache.rocketmq.acl.plain.PlainAccessResource.getRetryTopic;
+
+public class PlainAccessValidator implements AccessValidator {
+
+    private PlainAclPlugEngine aclPlugEngine;
+
+    public PlainAccessValidator() {
+        aclPlugEngine = new PlainAclPlugEngine();
+    }
+
+    @Override
+    public AccessResource parse(RemotingCommand request, String remoteAddr) {
+        PlainAccessResource accessResource = new PlainAccessResource();
+        accessResource.setRemoteAddr(remoteAddr);
+        accessResource.setRequestCode(request.getCode());
+        accessResource.setAccessKey(request.getExtFields().get(SessionCredentials.AccessKey));
+        accessResource.setSignature(request.getExtFields().get(SessionCredentials.Signature));
+        accessResource.setSecretToken(request.getExtFields().get(SessionCredentials.SecurityToken));
+
+        try {
+            // resource 和 permission 转换
+            switch (request.getCode()) {
+                case RequestCode.SEND_MESSAGE:
+                    accessResource.addResourceAndPerm(request.getExtFields().get("topic"), Permission.PUB);
+                    break;
+                case RequestCode.SEND_MESSAGE_V2:
+                    accessResource.addResourceAndPerm(request.getExtFields().get("b"), Permission.PUB);
+                    break;
+                case RequestCode.CONSUMER_SEND_MSG_BACK:
+                    accessResource.addResourceAndPerm(request.getExtFields().get("originTopic"), Permission.PUB);
+                    accessResource.addResourceAndPerm(getRetryTopic(request.getExtFields().get("group")), Permission.SUB);
+                    break;
+                case RequestCode.PULL_MESSAGE:
+                    accessResource.addResourceAndPerm(request.getExtFields().get("topic"), Permission.SUB);
+                    accessResource.addResourceAndPerm(getRetryTopic(request.getExtFields().get("consumerGroup")), Permission.SUB);
+                    break;
+                case RequestCode.QUERY_MESSAGE:
+                    accessResource.addResourceAndPerm(request.getExtFields().get("topic"), Permission.SUB);
+                    break;
+                case RequestCode.HEART_BEAT:
+                    HeartbeatData heartbeatData = HeartbeatData.decode(request.getBody(), HeartbeatData.class);
+                    for (ConsumerData data : heartbeatData.getConsumerDataSet()) {
+                        accessResource.addResourceAndPerm(getRetryTopic(data.getGroupName()), Permission.SUB);
+                        for (SubscriptionData subscriptionData: data.getSubscriptionDataSet()) {
+                            accessResource.addResourceAndPerm(subscriptionData.getTopic(), Permission.SUB);
+                        }
+                    }
+                    break;
+                case RequestCode.UNREGISTER_CLIENT:
+                    final UnregisterClientRequestHeader unregisterClientRequestHeader =
+                        (UnregisterClientRequestHeader) request
+                            .decodeCommandCustomHeader(UnregisterClientRequestHeader.class);
+                    accessResource.addResourceAndPerm(getRetryTopic(unregisterClientRequestHeader.getConsumerGroup()), Permission.SUB);
+                    break;
+                case RequestCode.GET_CONSUMER_LIST_BY_GROUP:
+                    final GetConsumerListByGroupRequestHeader getConsumerListByGroupRequestHeader =
+                        (GetConsumerListByGroupRequestHeader) request
+                            .decodeCommandCustomHeader(GetConsumerListByGroupRequestHeader.class);
+                    accessResource.addResourceAndPerm(getRetryTopic(getConsumerListByGroupRequestHeader.getConsumerGroup()), Permission.SUB);
+                    break;
+                case RequestCode.UPDATE_CONSUMER_OFFSET:
+                    final UpdateConsumerOffsetRequestHeader updateConsumerOffsetRequestHeader =
+                        (UpdateConsumerOffsetRequestHeader) request
+                            .decodeCommandCustomHeader(UpdateConsumerOffsetRequestHeader.class);
+                    accessResource.addResourceAndPerm(getRetryTopic(updateConsumerOffsetRequestHeader.getConsumerGroup()), Permission.SUB);
+                    accessResource.addResourceAndPerm(updateConsumerOffsetRequestHeader.getTopic(), Permission.SUB);
+                    break;
+                default:
+                    break;
+
+            }
+        } catch (Throwable t) {
+            throw new AclException(t.getMessage(), -1, t);
+        }
+
+
+        // content
+        SortedMap<String, String> map = new TreeMap<String, String>();
+        for (Map.Entry<String, String> entry : request.getExtFields().entrySet()) {
+            if (!SessionCredentials.Signature.equals(entry.getKey())) {
+                map.put(entry.getKey(), entry.getValue());
+            }
+        }
+        accessResource.setContent(AclUtils.combineRequestContent(request, map));
+
+        return accessResource;
+    }
+
+    @Override
+    public void validate(AccessResource accessResource) {
+        AuthenticationResult authenticationResult = null;
+        try {
+            authenticationResult = aclPlugEngine.eachCheckAuthentication((PlainAccessResource) accessResource);
+            if (authenticationResult.isSucceed())
+                return;
+        } catch (Exception e) {
+            throw new AclPlugRuntimeException(String.format("validate exception AccessResource data %s", accessResource.toString()), e);
+        }
+        if (authenticationResult.getException() != null) {
+            throw new AclPlugRuntimeException(String.format("eachCheck the inspection appear exception, accessControl data is %s", accessResource.toString()), authenticationResult.getException());
+        }
+        if (authenticationResult.getPlainAccessResource() != null || !authenticationResult.isSucceed()) {
+            throw new AclPlugRuntimeException(String.format("%s accessControl data is %s", authenticationResult.getResultString(), accessResource.toString()));
+        }
+    }
+
+}

acl/src/main/java/org/apache/rocketmq/acl/plug/PlainAclPlugEngine.java → acl/src/main/java/org/apache/rocketmq/acl/plain/PlainAclPlugEngine.java

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.rocketmq.acl.plug;
+package org.apache.rocketmq.acl.plain;
 
 import java.io.IOException;
 import java.lang.reflect.Field;
@@ -32,6 +32,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.acl.common.AclUtils;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.ServiceThread;
 import org.apache.rocketmq.common.constant.LoggerName;
@@ -123,9 +124,9 @@ public class PlainAclPlugEngine {
         }
     }
 
-    private void handleAccessControl(AccessControl accessControl) {
-        if (accessControl instanceof BrokerAccessControl) {
-            BrokerAccessControl brokerAccessControl = (BrokerAccessControl) accessControl;
+    private void handleAccessControl(PlainAccessResource plainAccessResource) {
+        if (plainAccessResource instanceof BrokerAccessControl) {
+            BrokerAccessControl brokerAccessControl = (BrokerAccessControl) plainAccessResource;
             if (brokerAccessControl.isAdmin()) {
                 brokerAccessControl.setUpdateAndCreateSubscriptiongroup(true);
                 brokerAccessControl.setDeleteSubscriptiongroup(true);
@@ -141,55 +142,55 @@ public class PlainAclPlugEngine {
         authenticationInfo = null;
     }
 
-    public void setAccessControl(AccessControl accessControl) throws AclPlugRuntimeException {
-        if (accessControl.getAccount() == null || accessControl.getPassword() == null
-            || accessControl.getAccount().length() <= 6 || accessControl.getPassword().length() <= 6) {
+    public void setAccessControl(PlainAccessResource plainAccessResource) throws AclPlugRuntimeException {
+        if (plainAccessResource.getAccessKey() == null || plainAccessResource.getSignature() == null
+            || plainAccessResource.getAccessKey().length() <= 6 || plainAccessResource.getSignature().length() <= 6) {
             throw new AclPlugRuntimeException(String.format(
                 "The account password cannot be null and is longer than 6, account is %s  password is %s",
-                accessControl.getAccount(), accessControl.getPassword()));
+                plainAccessResource.getAccessKey(), plainAccessResource.getSignature()));
         }
         try {
-            handleAccessControl(accessControl);
-            NetaddressStrategy netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(accessControl);
-            List<AuthenticationInfo> accessControlAddressList = accessControlMap.get(accessControl.getAccount());
+            handleAccessControl(plainAccessResource);
+            NetaddressStrategy netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(plainAccessResource);
+            List<AuthenticationInfo> accessControlAddressList = accessControlMap.get(plainAccessResource.getAccessKey());
             if (accessControlAddressList == null) {
                 accessControlAddressList = new ArrayList<>();
-                accessControlMap.put(accessControl.getAccount(), accessControlAddressList);
+                accessControlMap.put(plainAccessResource.getAccessKey(), accessControlAddressList);
             }
             AuthenticationInfo authenticationInfo = new AuthenticationInfo(
-                accessContralAnalysis.analysis(accessControl), accessControl, netaddressStrategy);
+                accessContralAnalysis.analysis(plainAccessResource), plainAccessResource, netaddressStrategy);
             accessControlAddressList.add(authenticationInfo);
             log.info("authenticationInfo is {}", authenticationInfo.toString());
         } catch (Exception e) {
             throw new AclPlugRuntimeException(
-                String.format("Exception info %s  %s", e.getMessage(), accessControl.toString()), e);
+                String.format("Exception info %s  %s", e.getMessage(), plainAccessResource.toString()), e);
         }
     }
 
-    public void setAccessControlList(List<AccessControl> accessControlList) throws AclPlugRuntimeException {
-        for (AccessControl accessControl : accessControlList) {
-            setAccessControl(accessControl);
+    public void setAccessControlList(List<PlainAccessResource> plainAccessResourceList) throws AclPlugRuntimeException {
+        for (PlainAccessResource plainAccessResource : plainAccessResourceList) {
+            setAccessControl(plainAccessResource);
         }
     }
 
-    public void setNetaddressAccessControl(AccessControl accessControl) throws AclPlugRuntimeException {
+    public void setNetaddressAccessControl(PlainAccessResource plainAccessResource) throws AclPlugRuntimeException {
         try {
-            authenticationInfo = new AuthenticationInfo(accessContralAnalysis.analysis(accessControl), accessControl, netaddressStrategyFactory.getNetaddressStrategy(accessControl));
+            authenticationInfo = new AuthenticationInfo(accessContralAnalysis.analysis(plainAccessResource), plainAccessResource, netaddressStrategyFactory.getNetaddressStrategy(plainAccessResource));
             log.info("default authenticationInfo is {}", authenticationInfo.toString());
         } catch (Exception e) {
-            throw new AclPlugRuntimeException(accessControl.toString(), e);
+            throw new AclPlugRuntimeException(plainAccessResource.toString(), e);
         }
 
     }
 
-    public AuthenticationInfo getAccessControl(AccessControl accessControl) {
-        if (accessControl.getAccount() == null && authenticationInfo != null) {
-            return authenticationInfo.getNetaddressStrategy().match(accessControl) ? authenticationInfo : null;
+    public AuthenticationInfo getAccessControl(PlainAccessResource plainAccessResource) {
+        if (plainAccessResource.getAccessKey() == null && authenticationInfo != null) {
+            return authenticationInfo.getNetaddressStrategy().match(plainAccessResource) ? authenticationInfo : null;
         } else {
-            List<AuthenticationInfo> accessControlAddressList = accessControlMap.get(accessControl.getAccount());
+            List<AuthenticationInfo> accessControlAddressList = accessControlMap.get(plainAccessResource.getAccessKey());
             if (accessControlAddressList != null) {
                 for (AuthenticationInfo ai : accessControlAddressList) {
-                    if (ai.getNetaddressStrategy().match(accessControl) && ai.getAccessControl().getPassword().equals(accessControl.getPassword())) {
+                    if (ai.getNetaddressStrategy().match(plainAccessResource) && ai.getPlainAccessResource().getSignature().equals(plainAccessResource.getSignature())) {
                         return ai;
                     }
                 }
@@ -198,15 +199,15 @@ public class PlainAclPlugEngine {
         return null;
     }
 
-    public AuthenticationResult eachCheckAuthentication(AccessControl accessControl) {
+    public AuthenticationResult eachCheckAuthentication(PlainAccessResource plainAccessResource) {
         AuthenticationResult authenticationResult = new AuthenticationResult();
-        AuthenticationInfo authenticationInfo = getAccessControl(accessControl);
+        AuthenticationInfo authenticationInfo = getAccessControl(plainAccessResource);
         if (authenticationInfo != null) {
-            boolean boo = authentication(authenticationInfo, accessControl, authenticationResult);
+            boolean boo = authentication(authenticationInfo, plainAccessResource, authenticationResult);
             authenticationResult.setSucceed(boo);
-            authenticationResult.setAccessControl(authenticationInfo.getAccessControl());
+            authenticationResult.setPlainAccessResource(authenticationInfo.getPlainAccessResource());
         } else {
-            authenticationResult.setResultString("accessControl is null, Please check login, password, IP\"");
+            authenticationResult.setResultString("plainAccessResource is null, Please check login, password, IP\"");
         }
         return authenticationResult;
     }
@@ -226,18 +227,18 @@ public class PlainAclPlugEngine {
         }
     }
 
-    public boolean authentication(AuthenticationInfo authenticationInfo, AccessControl accessControl,
+    public boolean authentication(AuthenticationInfo authenticationInfo, PlainAccessResource plainAccessResource,
         AuthenticationResult authenticationResult) {
-        int code = accessControl.getCode();
+        int code = plainAccessResource.getRequestCode();
         if (!authenticationInfo.getAuthority().get(code)) {
             authenticationResult.setResultString(String.format("code is %d Authentication failed", code));
             return false;
         }
-        if (!(authenticationInfo.getAccessControl() instanceof BrokerAccessControl)) {
+        if (!(authenticationInfo.getPlainAccessResource() instanceof BrokerAccessControl)) {
             return true;
         }
-        BrokerAccessControl borker = (BrokerAccessControl) authenticationInfo.getAccessControl();
-        String topicName = accessControl.getTopic();
+        BrokerAccessControl borker = (BrokerAccessControl) authenticationInfo.getPlainAccessResource();
+        String topicName = plainAccessResource.getTopic();
         if (code == 10 || code == 310 || code == 320) {
             if (borker.getPermitSendTopic().contains(topicName)) {
                 return true;
@@ -284,8 +285,8 @@ public class PlainAclPlugEngine {
             }
         }
 
-        public Map<Integer, Boolean> analysis(AccessControl accessControl) {
-            Class<? extends AccessControl> clazz = accessControl.getClass();
+        public Map<Integer, Boolean> analysis(PlainAccessResource plainAccessResource) {
+            Class<? extends PlainAccessResource> clazz = plainAccessResource.getClass();
             Map<Integer, Field> codeAndField = classTocodeAndMentod.get(clazz);
             if (codeAndField == null) {
                 codeAndField = new HashMap<>();
@@ -305,8 +306,8 @@ public class PlainAclPlugEngine {
 
                 }
                 if (codeAndField.isEmpty()) {
-                    throw new AclPlugRuntimeException(String.format("AccessControl nonexistent code , name %s",
-                        accessControl.getClass().getName()));
+                    throw new AclPlugRuntimeException(String.format("PlainAccessResource nonexistent code , name %s",
+                        plainAccessResource.getClass().getName()));
                 }
                 classTocodeAndMentod.put(clazz, codeAndField);
             }
@@ -315,11 +316,11 @@ public class PlainAclPlugEngine {
             try {
                 while (it.hasNext()) {
                     Entry<Integer, Field> e = it.next();
-                    authority.put(e.getKey(), (Boolean) e.getValue().get(accessControl));
+                    authority.put(e.getKey(), (Boolean) e.getValue().get(plainAccessResource));
                 }
             } catch (IllegalArgumentException | IllegalAccessException e) {
                 throw new AclPlugRuntimeException(
-                    String.format("analysis on failure AccessControl is %s", AccessControl.class.getName()), e);
+                    String.format("analysis on failure PlainAccessResource is %s", PlainAccessResource.class.getName()), e);
             }
             return authority;
         }

acl/src/test/java/org/apache/rocketmq/acl/plug/AclUtilsTest.java → acl/src/test/java/org/apache/rocketmq/acl/plain/AclUtilsTest.java

@@ -14,11 +14,12 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.rocketmq.acl.plug;
+package org.apache.rocketmq.acl.plain;
 
 import java.util.ArrayList;
 import java.util.List;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.acl.common.AclUtils;
 import org.junit.Assert;
 import org.junit.Test;
 

acl/src/test/java/org/apache/rocketmq/acl/plug/NetaddressStrategyTest.java → acl/src/test/java/org/apache/rocketmq/acl/plain/NetaddressStrategyTest.java

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.rocketmq.acl.plug;
+package org.apache.rocketmq.acl.plain;
 
 import org.junit.Assert;
 import org.junit.Test;
@@ -25,111 +25,111 @@ public class NetaddressStrategyTest {
 
     @Test
     public void NetaddressStrategyFactoryTest() {
-        AccessControl accessControl = new AccessControl();
-        NetaddressStrategy netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(accessControl);
+        PlainAccessResource plainAccessResource = new PlainAccessResource();
+        NetaddressStrategy netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(plainAccessResource);
         Assert.assertEquals(netaddressStrategy, NetaddressStrategyFactory.NULL_NET_ADDRESS_STRATEGY);
 
-        accessControl.setNetaddress("*");
-        netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(accessControl);
+        plainAccessResource.setRemoteAddr("*");
+        netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(plainAccessResource);
         Assert.assertEquals(netaddressStrategy, NetaddressStrategyFactory.NULL_NET_ADDRESS_STRATEGY);
 
-        accessControl.setNetaddress("127.0.0.1");
-        netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(accessControl);
+        plainAccessResource.setRemoteAddr("127.0.0.1");
+        netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(plainAccessResource);
         Assert.assertEquals(netaddressStrategy.getClass(), NetaddressStrategyFactory.OneNetaddressStrategy.class);
 
-        accessControl.setNetaddress("127.0.0.1,127.0.0.2,127.0.0.3");
-        netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(accessControl);
+        plainAccessResource.setRemoteAddr("127.0.0.1,127.0.0.2,127.0.0.3");
+        netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(plainAccessResource);
         Assert.assertEquals(netaddressStrategy.getClass(), NetaddressStrategyFactory.MultipleNetaddressStrategy.class);
 
-        accessControl.setNetaddress("127.0.0.{1,2,3}");
-        netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(accessControl);
+        plainAccessResource.setRemoteAddr("127.0.0.{1,2,3}");
+        netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(plainAccessResource);
         Assert.assertEquals(netaddressStrategy.getClass(), NetaddressStrategyFactory.MultipleNetaddressStrategy.class);
 
-        accessControl.setNetaddress("127.0.0.1-200");
-        netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(accessControl);
+        plainAccessResource.setRemoteAddr("127.0.0.1-200");
+        netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(plainAccessResource);
         Assert.assertEquals(netaddressStrategy.getClass(), NetaddressStrategyFactory.RangeNetaddressStrategy.class);
 
-        accessControl.setNetaddress("127.0.0.*");
-        netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(accessControl);
+        plainAccessResource.setRemoteAddr("127.0.0.*");
+        netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(plainAccessResource);
         Assert.assertEquals(netaddressStrategy.getClass(), NetaddressStrategyFactory.RangeNetaddressStrategy.class);
 
-        accessControl.setNetaddress("127.0.1-20.*");
-        netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(accessControl);
+        plainAccessResource.setRemoteAddr("127.0.1-20.*");
+        netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(plainAccessResource);
         Assert.assertEquals(netaddressStrategy.getClass(), NetaddressStrategyFactory.RangeNetaddressStrategy.class);
     }
 
     @Test(expected = AclPlugRuntimeException.class)
     public void verifyTest() {
-        AccessControl accessControl = new AccessControl();
-        accessControl.setNetaddress("127.0.0.1");
-        netaddressStrategyFactory.getNetaddressStrategy(accessControl);
-        accessControl.setNetaddress("256.0.0.1");
-        netaddressStrategyFactory.getNetaddressStrategy(accessControl);
+        PlainAccessResource plainAccessResource = new PlainAccessResource();
+        plainAccessResource.setRemoteAddr("127.0.0.1");
+        netaddressStrategyFactory.getNetaddressStrategy(plainAccessResource);
+        plainAccessResource.setRemoteAddr("256.0.0.1");
+        netaddressStrategyFactory.getNetaddressStrategy(plainAccessResource);
     }
 
     @Test
     public void nullNetaddressStrategyTest() {
-        boolean isMatch = NetaddressStrategyFactory.NULL_NET_ADDRESS_STRATEGY.match(new AccessControl());
+        boolean isMatch = NetaddressStrategyFactory.NULL_NET_ADDRESS_STRATEGY.match(new PlainAccessResource());
         Assert.assertTrue(isMatch);
     }
 
     public void oneNetaddressStrategyTest() {
-        AccessControl accessControl = new AccessControl();
-        accessControl.setNetaddress("127.0.0.1");
-        NetaddressStrategy netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(accessControl);
-        accessControl.setNetaddress("");
-        boolean match = netaddressStrategy.match(accessControl);
+        PlainAccessResource plainAccessResource = new PlainAccessResource();
+        plainAccessResource.setRemoteAddr("127.0.0.1");
+        NetaddressStrategy netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(plainAccessResource);
+        plainAccessResource.setRemoteAddr("");
+        boolean match = netaddressStrategy.match(plainAccessResource);
         Assert.assertFalse(match);
 
-        accessControl.setNetaddress("127.0.0.2");
-        match = netaddressStrategy.match(accessControl);
+        plainAccessResource.setRemoteAddr("127.0.0.2");
+        match = netaddressStrategy.match(plainAccessResource);
         Assert.assertFalse(match);
 
-        accessControl.setNetaddress("127.0.0.1");
-        match = netaddressStrategy.match(accessControl);
+        plainAccessResource.setRemoteAddr("127.0.0.1");
+        match = netaddressStrategy.match(plainAccessResource);
         Assert.assertTrue(match);
     }
 
     @Test
     public void multipleNetaddressStrategyTest() {
-        AccessControl accessControl = new AccessControl();
-        accessControl.setNetaddress("127.0.0.1,127.0.0.2,127.0.0.3");
-        NetaddressStrategy netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(accessControl);
+        PlainAccessResource plainAccessResource = new PlainAccessResource();
+        plainAccessResource.setRemoteAddr("127.0.0.1,127.0.0.2,127.0.0.3");
+        NetaddressStrategy netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(plainAccessResource);
         multipleNetaddressStrategyTest(netaddressStrategy);
 
-        accessControl.setNetaddress("127.0.0.{1,2,3}");
-        netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(accessControl);
+        plainAccessResource.setRemoteAddr("127.0.0.{1,2,3}");
+        netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(plainAccessResource);
         multipleNetaddressStrategyTest(netaddressStrategy);
 
     }
 
     @Test(expected = AclPlugRuntimeException.class)
     public void multipleNetaddressStrategyExceptionTest() {
-        AccessControl accessControl = new AccessControl();
-        accessControl.setNetaddress("127.0.0.1,2,3}");
-        netaddressStrategyFactory.getNetaddressStrategy(accessControl);
+        PlainAccessResource plainAccessResource = new PlainAccessResource();
+        plainAccessResource.setRemoteAddr("127.0.0.1,2,3}");
+        netaddressStrategyFactory.getNetaddressStrategy(plainAccessResource);
     }
 
     private void multipleNetaddressStrategyTest(NetaddressStrategy netaddressStrategy) {
-        AccessControl accessControl = new AccessControl();
-        accessControl.setNetaddress("127.0.0.1");
-        boolean match = netaddressStrategy.match(accessControl);
+        PlainAccessResource plainAccessResource = new PlainAccessResource();
+        plainAccessResource.setRemoteAddr("127.0.0.1");
+        boolean match = netaddressStrategy.match(plainAccessResource);
         Assert.assertTrue(match);
 
-        accessControl.setNetaddress("127.0.0.2");
-        match = netaddressStrategy.match(accessControl);
+        plainAccessResource.setRemoteAddr("127.0.0.2");
+        match = netaddressStrategy.match(plainAccessResource);
         Assert.assertTrue(match);
 
-        accessControl.setNetaddress("127.0.0.3");
-        match = netaddressStrategy.match(accessControl);
+        plainAccessResource.setRemoteAddr("127.0.0.3");
+        match = netaddressStrategy.match(plainAccessResource);
         Assert.assertTrue(match);
 
-        accessControl.setNetaddress("127.0.0.4");
-        match = netaddressStrategy.match(accessControl);
+        plainAccessResource.setRemoteAddr("127.0.0.4");
+        match = netaddressStrategy.match(plainAccessResource);
         Assert.assertFalse(match);
 
-        accessControl.setNetaddress("127.0.0.0");
-        match = netaddressStrategy.match(accessControl);
+        plainAccessResource.setRemoteAddr("127.0.0.0");
+        match = netaddressStrategy.match(plainAccessResource);
         Assert.assertFalse(match);
 
     }
@@ -137,25 +137,25 @@ public class NetaddressStrategyTest {
     @Test
     public void rangeNetaddressStrategyTest() {
         String head = "127.0.0.";
-        AccessControl accessControl = new AccessControl();
-        accessControl.setNetaddress("127.0.0.1-200");
-        NetaddressStrategy netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(accessControl);
+        PlainAccessResource plainAccessResource = new PlainAccessResource();
+        plainAccessResource.setRemoteAddr("127.0.0.1-200");
+        NetaddressStrategy netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(plainAccessResource);
         rangeNetaddressStrategyTest(netaddressStrategy, head, 1, 200, true);
-        accessControl.setNetaddress("127.0.0.*");
-        netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(accessControl);
+        plainAccessResource.setRemoteAddr("127.0.0.*");
+        netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(plainAccessResource);
         rangeNetaddressStrategyTest(netaddressStrategy, head, 0, 255, true);
 
-        accessControl.setNetaddress("127.0.1-200.*");
-        netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(accessControl);
+        plainAccessResource.setRemoteAddr("127.0.1-200.*");
+        netaddressStrategy = netaddressStrategyFactory.getNetaddressStrategy(plainAccessResource);
         rangeNetaddressStrategyThirdlyTest(netaddressStrategy, head, 1, 200);
     }
 
     private void rangeNetaddressStrategyTest(NetaddressStrategy netaddressStrategy, String head, int start, int end,
         boolean isFalse) {
-        AccessControl accessControl = new AccessControl();
+        PlainAccessResource plainAccessResource = new PlainAccessResource();
         for (int i = -10; i < 300; i++) {
-            accessControl.setNetaddress(head + i);
-            boolean match = netaddressStrategy.match(accessControl);
+            plainAccessResource.setRemoteAddr(head + i);
+            boolean match = netaddressStrategy.match(plainAccessResource);
             if (isFalse && i >= start && i <= end) {
                 Assert.assertTrue(match);
                 continue;
@@ -192,9 +192,9 @@ public class NetaddressStrategyTest {
     }
 
     private void rangeNetaddressStrategyExceptionTest(String netaddress) {
-        AccessControl accessControl = new AccessControl();
-        accessControl.setNetaddress(netaddress);
-        netaddressStrategyFactory.getNetaddressStrategy(accessControl);
+        PlainAccessResource plainAccessResource = new PlainAccessResource();
+        plainAccessResource.setRemoteAddr(netaddress);
+        netaddressStrategyFactory.getNetaddressStrategy(plainAccessResource);
     }
 
 }

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

@@ -14,22 +14,52 @@
 #  limitations under the License.
 
 onlyNetAddress:
-  netaddress: 10.10.103.*
+  remoteAddr: 10.10.103.*
   noPermitPullTopic:
   - broker-a
 
 list:
-- account: RocketMQ
-  password: 1234567
-  netaddress: 192.0.0.*
+- accessKey: RocketMQ
+  signature: 1234567
+  remoteAddr: 192.0.0.*
   admin: true
   permitSendTopic:
   - test1
   - test2
-- account: RocketMQ
-  password: 1234567
-  netaddress: 192.0.2.1
+- accessKey: RocketMQ
+  signature: 1234567
+  remoteAddr: 192.0.2.1
   permitSendTopic:
   - test3
   - test4
-    
\ No newline at end of file
+
+
+## suggested format
+
+globalWhiteRemoteAddresses:
+  - 10.10.103.*
+  - 192.168.0.*
+
+accounts:
+- accessKey: ak1
+  secretKey: sk1
+  whiteRemoteAddress: 192.168.0.*
+  admin: false
+  defaultTopicPerm: DENY
+  defaultGroupPerm: SUB
+  topics:
+  - topicA=DENY
+  - topicB=PUB|SUB
+  - topicC=SUB
+  groups:
+  # the group should convert to retry topic
+  - groupA=DENY
+  - groupB=SUB
+  - groupC=SUB
+
+- accessKey: ak2
+  secretKey: sk2
+  whiteRemoteAddress: 192.168.1.*
+  # if it is admin, it could access all resources
+  admin: true
+

distribution/conf/transport.yml

@@ -14,20 +14,20 @@
 #  limitations under the License.
 
 onlyNetAddress:
-  netaddress: 192.168.0.*
+  remoteAddr: 192.168.0.*
   noPermitPullTopic:
     - broker-a
 
 list:
   - account: RocketMQ
-    password: 1234567
-    netaddress: 192.168.0.*
+    signature: 1234567
+    remoteAddr: 192.168.0.*
     permitSendTopic:
       - TopicTest
       - test2
   - account: RocketMQ
-    password: 1234567
-    netaddress: 192.168.2.1
+    signature: 1234567
+    remoteAddr: 192.168.2.1
     permitSendTopic:
       - test3
       - test4

pom.xml

@@ -593,6 +593,11 @@
                 <artifactId>snakeyaml</artifactId>
                 <version>1.19</version>
             </dependency>
+            <dependency>
+                <groupId>commons-codec</groupId>
+                <artifactId>commons-codec</artifactId>
+                <version>1.9</version>
+            </dependency>
             <dependency>
                 <groupId>org.apache.logging.log4j</groupId>
                 <artifactId>log4j-core</artifactId>