Merge 6574b69 into 28e4ac5

.gitignore

@@ -12,3 +12,4 @@ devenv
 !LICENSE-BIN
 .DS_Store
 localbin
+nohup.out

README.md

@@ -38,7 +38,6 @@ It offers a variety of features:
 
 ## Apache RocketMQ Community
 * [RocketMQ Community Projects](https://github.com/apache/rocketmq-externals)
-
 ----------
 
 ## Contributing

acl/pom.xml

@@ -0,0 +1,74 @@
+<!-- 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. -->
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
+         xmlns="http://maven.apache.org/POM/4.0.0">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.rocketmq</groupId>
+        <artifactId>rocketmq-all</artifactId>
+        <version>4.4.0-SNAPSHOT</version>
+    </parent>
+    <artifactId>rocketmq-acl</artifactId>
+    <name>rocketmq-acl ${project.version}</name>
+
+    <url>http://maven.apache.org</url>
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    </properties>
+    <dependencies>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>rocketmq-remoting</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>rocketmq-logging</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>rocketmq-common</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>rocketmq-srvutil</artifactId>
+        </dependency>
+        <dependency>
+            <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>
+        </dependency>
+
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-classic</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+    </dependencies>
+</project>

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

@@ -0,0 +1,21 @@
+/*
+ * 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;
+
+public interface AccessResource {
+}

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

@@ -0,0 +1,38 @@
+/*
+ * 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 org.apache.rocketmq.remoting.protocol.RemotingCommand;
+
+public interface AccessValidator {
+    /**
+     * Parse to get the AccessResource(user, resource, needed permission)
+     *
+     * @param request
+     * @param remoteAddr
+     * @return Plain access resource result,include access key,signature and some other access attributes.
+     */
+    AccessResource parse(RemotingCommand request, String remoteAddr);
+
+    /**
+     * Validate the access resource.
+     *
+     * @param accessResource
+     */
+    void validate(AccessResource accessResource);
+}

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

@@ -0,0 +1,98 @@
+/*
+ * 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.common;
+
+import java.lang.reflect.Field;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.rocketmq.remoting.CommandCustomHeader;
+import org.apache.rocketmq.remoting.RPCHook;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+
+import static org.apache.rocketmq.acl.common.SessionCredentials.ACCESS_KEY;
+import static org.apache.rocketmq.acl.common.SessionCredentials.SECURITY_TOKEN;
+import static org.apache.rocketmq.acl.common.SessionCredentials.SIGNATURE;
+
+public class AclClientRPCHook implements RPCHook {
+    private final SessionCredentials sessionCredentials;
+    protected ConcurrentHashMap<Class<? extends CommandCustomHeader>, Field[]> fieldCache =
+        new ConcurrentHashMap<Class<? extends CommandCustomHeader>, Field[]>();
+
+    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(ACCESS_KEY, sessionCredentials.getAccessKey());
+
+        // The SecurityToken value is unneccessary,user can choose this one.
+        if (sessionCredentials.getSecurityToken() != null) {
+            request.addExtField(SECURITY_TOKEN, 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(ACCESS_KEY, ak);
+        if (securityToken != null) {
+            map.put(SECURITY_TOKEN, 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

@@ -0,0 +1,66 @@
+/*
+ * 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.common;
+
+public class AclException extends RuntimeException {
+    private static final long serialVersionUID = -7256002576788700354L;
+
+    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 message) {
+        super(message);
+    }
+
+    public AclException(String message, Throwable throwable) {
+        super(message, throwable);
+    }
+
+    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;
+    }
+}