聊聊rocketmq的AclClientRPCHook

序

本文主要研究一下 rocketmq 的 AclClientRPCHook

RPCHook

rocketmq-remoting-4.5.2-sources.jar!/org/apache/rocketmq/remoting/RPCHook.java

public interface RPCHook {void doBeforeRequest(final String remoteAddr, final RemotingCommand request);

    void doAfterResponse(final String remoteAddr, final RemotingCommand request,
        final RemotingCommand response);
}

• RPCHook 定义了 doBeforeRequest、doAfterResponse 方法

AclClientRPCHook

rocketmq-acl-4.5.2-sources.jar!/org/apache/rocketmq/acl/common/AclClientRPCHook.java

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;}
}

• AclClientRPCHook 实现了 RPCHook 接口，其构造器接收 SessionCredentials 参数；其 doBeforeRequest 首先通过 parseRequestContent 从 request 读取 CommandCustomHeader，将其 field 连同 accessKey、securityToken 放到一个 SortedMap，再通过 AclUtils.combineRequestContent 计算要发送的请求内容；然后通过 AclUtils.calSignature 计算出 signature，最后往 request 的 extFields 添加 SIGNATURE、ACCESS_KEY；若设置 securityToken，则会往 request 的 extFields 添加 SECURITY_TOKEN

SessionCredentials

rocketmq-acl-4.5.2-sources.jar!/org/apache/rocketmq/acl/common/SessionCredentials.java

public class SessionCredentials {public static final Charset CHARSET = Charset.forName("UTF-8");
    public static final String ACCESS_KEY = "AccessKey";
    public static final String SECRET_KEY = "SecretKey";
    public static final String SIGNATURE = "Signature";
    public static final String SECURITY_TOKEN = "SecurityToken";

    public static final String KEY_FILE = System.getProperty("rocketmq.client.keyFile",
        System.getProperty("user.home") + File.separator + "key");

    private String accessKey;
    private String secretKey;
    private String securityToken;
    private String signature;

    public SessionCredentials() {
        String keyContent = null;
        try {keyContent = MixAll.file2String(KEY_FILE);
        } 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(ACCESS_KEY);
            if (value != null) {this.accessKey = value.trim();
            }
        }
        {String value = prop.getProperty(SECRET_KEY);
            if (value != null) {this.secretKey = value.trim();
            }
        }
        {String value = prop.getProperty(SECURITY_TOKEN);
            if (value != null) {this.securityToken = value.trim();
            }
        }
    }

    //......
}

• SessionCredentials 提供了三个构造器，一个无参构造器从 KEY_FILE 加载 keyContent 然后解析为 Properties 再通过 updateContent 方法给 accessKey、secretKey、securityToken 赋值；一个是 accessKey, secretKey 的构造器；还有一个是 accessKey, secretKey, securityToken 的构造器

AclUtils

rocketmq-acl-4.5.2-sources.jar!/org/apache/rocketmq/acl/common/AclUtils.java

public class AclUtils {private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.COMMON_LOGGER_NAME);

    public static byte[] combineRequestContent(RemotingCommand request, SortedMap<String, String> fieldsMap) {
        try {StringBuilder sb = new StringBuilder("");
            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;
    }

    //......
}

• combineRequestContent 首先将 fieldsMap 拼接为字符串，然后通过 AclUtils.combineBytes 将其与 request.getBody() 结合在一起；calSignature 方法内部是委托给 AclSigner.calSignature(data, secretKey) 来实现

AclSigner

rocketmq-acl-4.5.2-sources.jar!/org/apache/rocketmq/acl/common/AclSigner.java

public class AclSigner {public static final Charset DEFAULT_CHARSET = Charset.forName("UTF-8");
    public static final SigningAlgorithm DEFAULT_ALGORITHM = 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, DEFAULT_ALGORITHM, DEFAULT_CHARSET);
    }

    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), DEFAULT_CHARSET);
        } 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, DEFAULT_ALGORITHM, DEFAULT_CHARSET);
    }

    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), DEFAULT_CHARSET);
        } 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);
        }
    }

}

• calSignature 默认使用的是 SigningAlgorithm.HmacSHA1 及 Charset.forName(“UTF-8”) 字符集来签名；signAndBase64Encode 方法首先通过 sign 方法签名，然后将其转为 Base64 的字符串

小结

AclClientRPCHook 实现了 RPCHook 接口，其构造器接收 SessionCredentials 参数；其 doBeforeRequest 首先通过 parseRequestContent 从 request 读取 CommandCustomHeader，将其 field 连同 accessKey、securityToken 放到一个 SortedMap，再通过 AclUtils.combineRequestContent 计算要发送的请求内容；然后通过 AclUtils.calSignature 计算出 signature，最后往 request 的 extFields 添加 SIGNATURE、ACCESS_KEY；若设置 securityToken，则会往 request 的 extFields 添加 SECURITY_TOKEN

doc

• AclClientRPCHook