RocketMQ ACL架构升级的一些建议与实践

Posted 2023-03-06 中间件兴趣圈

大家好，我是威哥，《RocketMQ技术内幕》、《RocketMQ实战》作者、RocketMQ社区首席布道师、极客时间《中间件核心技术与实战》专栏作者、中通快递基础架构资深架构师，越努力越幸运，唯有坚持不懈，与大家共勉。

随着业务的高速发展，消息中间件几乎涵盖了全司所有链路，如何打造安全可靠、性能达标的消息中间件成为了一个非常紧迫并具有较大的挑战性。
作为我们公司消息中间件第一责任人，这个任务自然而然的落到了我的身上。

接到这个任务，其实我一点都不慌，因为我知道RocketMQ在4.4.0版本就开始引入了ACL机制，于是我一顿操作猛如虎，马上就提出了基于官方ACL的设计理念，再结合公司独有的账号授权体系，提出了一版本ACL设计方案，马上提交给我领导进行评审，在我满怀信息快速过了一遍方案时，**领导微微一笑，每一条消息都需要进行鉴权一次？中通这个消息体量，服务端能否承载这么大的负荷呢？你看能否这样这样。。。**听后，立马给我醍醐灌顶之效，意识到这是一个非常不错的点子？是什么呢？请听我慢慢道来。

1、RocketMQ ACL设计原理

RocketMQ ACL的设计原理如下图所示：

其核心的实现要点：

• 客户端在消息发送或消息消费时，客户端首先需要封装身份信息（访问者账号），并且为了避免密钥在网络中传输，需要在客户端对请求中的参数（例如消息属性）中的key先进行排序并拼接成一个字符串，然后使用该账号对应的密钥与拼接后的字符串进行md5签名，然后将生成后的密钥传递到服务端
• 服务端在接受信息后，从请求中解析出账号，并根据账号查找出对应的密钥，然后对参数采取客户端相同的策略生成密钥，如果服务端生成的密钥与客户端生成的一样，说明身份校验成功，否则校验失败，直接拒绝本次操作。
• 在身份校验成功后，再进行访问权限校验(ACL),也就是说验证用户的权限，即判断用户是有拥有资源(主题、消费组)的发送，获取查询权限。

从这里也有看出，RocketMQ的ACL机制一个非常显著的特点：身份校验+访问权限这两个操作融合在一起。

这样一个明显的弊端就是每一次请求，都需要进行重复的身份验证，向中通这样**万亿级消息流转（消息发送/消息消费）**的场景显然是不合适的，所以我们领导就提出了另外一个设想：能否基于连接级别的身份验证呢？也就是一个客户端只在建立连接时进行一次身份认证，后续通过这条连接发送/消费消息，都不需要再重复校验，只需要进行访问权限校验，这样将能极大的降低服务器的CPU的消耗，性能才能得以保证。

本文并不打算详细介绍RocketMQ ACL相关的知识，如果大家有兴趣，可以看我关于ACL的往期文章：

RocketMQ ACL使用指南

源码分析RocketMQ ACL实现机制

(架构实战)你的RocketMQ集群是安全的吗？

2、基于连接鉴权

基于连接鉴权，这是一个非常cool的点子，那如何实现呢？

经过调研，并参考了Kafka的ACL实现校验，发现了一个非常灵活的权限校验框架，并且已经集成在了java sdk中，那就是sasl，这个框架的包为javax.security.sasl包中。

网上关于sasl的文章众多，在这里我也不再重复，我在这里主要解决一个问题，就是我们在阅读完网上相关资料后，如何在代码层面进行落实，并正确理解sasl协议，以便大家在nio或者netty中引入sasl身份校验机制，从而解决分布式环境下的身份认证与访问控制。

在进行原理解读之前，我先放一段单机版的sasl示例代码，代码如下：

/** -----------------AuthMain-------------------*/
package com.vhcool.test.sasl;
import com.sun.tools.javac.util.Assert;
import javax.security.sasl.Sasl;
import javax.security.sasl.SaslClient;
import javax.security.sasl.SaslException;
import javax.security.sasl.SaslServer;
import java.util.HashMap;
import java.util.Map;
/**
 * <p> </p>
 *
 * @author lidawei
 * @date 2023/2/7
 * @since 1.0.0
 **/
public class AuthMain 
	public static void main(String[] args) throws SaslException 
		//服务端
		Map<String, String> props = new HashMap<>();
		props.put(Sasl.QOP, "auth");
		SaslServer ss = Sasl.createSaslServer("DIGEST-MD5", "xmpp", "myServer",
				props, new ServerCallbackHandler());
		//客户端
		SaslClient sc = Sasl.createSaslClient(new String[]"DIGEST-MD5",
				"tony", "xmpp", "myServer", null, new ClientCallbackHandler());
		byte[] challenge;
		byte[] response;
		//服务发出挑战
		challenge = ss.evaluateResponse(new byte[0]);
		//客户端做出回应
		response = sc.evaluateChallenge(challenge);
		//服务发出挑战
		challenge = ss.evaluateResponse(response);
		//客户端做出回应
		response = sc.evaluateChallenge(challenge);
		Assert.check(ss.isComplete());
		Assert.check(sc.isComplete());
	


/** -----------------ClientCallbackHandler-------------------*/
package com.vhcool.test.sasl;
import javax.security.auth.callback.*;
import javax.security.sasl.RealmCallback;

/**
 * <p> </p>
 *
 * @author lidawei
 * @date 2023/2/7
 * @since 1.0.0
 **/
public class ClientCallbackHandler implements CallbackHandler 
	@Override
	public void handle(Callback[] callbacks) 
		for (Callback cb : callbacks) 
			if (cb instanceof NameCallback) 
				NameCallback nc = (NameCallback) cb;
				//Collect username in application-specific manner
				nc.setName("username");
			 else if (cb instanceof PasswordCallback) 
				PasswordCallback pc = (PasswordCallback) cb;
				//Collect password in application-specific manner
				pc.setPassword("password".toCharArray());
			 else if (cb instanceof RealmCallback) 
				RealmCallback rc = (RealmCallback) cb;
				//Collect realm data in application-specific manner
				rc.setText("myServer");
			
		
	

/** -----------------ServerCallbackHandler-------------------*/
import javax.security.auth.callback.*;
import javax.security.sasl.AuthorizeCallback;
import javax.security.sasl.RealmCallback;
import java.io.IOException;
/**
 * <p> </p>
 *
 * @author lidawei
 * @date 2023/2/7
 * @since 1.0.0
 **/
public class ServerCallbackHandler implements CallbackHandler 
	@Override
	public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException 
		for (Callback cb : callbacks) 
			if (cb instanceof AuthorizeCallback) 
				AuthorizeCallback ac = (AuthorizeCallback) cb;
				//Perform application-specific authorization action
				ac.setAuthorized(true);
			 else if (cb instanceof NameCallback) 
				NameCallback nc = (NameCallback) cb;
				//Collect username in application-specific manner
				nc.setName("username");
			 else if (cb instanceof PasswordCallback) 
				PasswordCallback pc = (PasswordCallback) cb;
				//Collect password in application-specific manner
				pc.setPassword("password2".toCharArray());
			 else if (cb instanceof RealmCallback) 
				RealmCallback rc = (RealmCallback) cb;
				//Collect realm data in application-specific manner
				rc.setText("myServer");
			
		
	

Sasl协议的核心要点如下图所示：

核心要点都展示在图上了，这里最后再补充一下如何判断是否成功。

服务端发起挑战，客户端会根据服务端生成的挑战数据，在客户端进行挑战，再传递给服务端进行调整，服务端再次挑战后，如果返回值为空，并且isComplete方法返回ture，表示服务端鉴权成功，如果为空但isComplete方法返回fasle，表示鉴权失败。同样客户端收到服务端的挑战数据后，如果不为空，则需要继续调用挑战(evaluateResponse)方法,如果结果为空，并且isÇomplete方法为true表示成功，如果生成的挑战结果不为空，继续发送给服务端。

也就是这里的挑战轮次(次数)并不确定。

由于篇幅的问题，本文就不介绍到这里了，在这里只是介绍了单机版的实现，并且在原理途中如果采用的是分布式通信，哪些信息是需要通过网络传输的，那我们接下来的任务，就是如何在RocketMQ中引入sasl，由于RocketMQ的网络通信底层采用的是Netty框架，那如何在Netty框架中引入sasl呢？大家可以自己先行思考，我将会在后续的文章中进行公开发表，敬请期待。

本文首发：https://www.codingw.net/Article?id=784

---

见字如面,我是威哥,一个从普通二本院校毕业,从未曾接触分布式、微服务、高并发到通过技术分享实现职场蜕变，成长为RocketMQ社区优秀布道师、大厂资深架构师，出版《RocketMQ技术内幕》、《RocketMQ实战》两本技术书籍，在CSDN中记录了我的成长历程，欢迎大家关注，私信，一起交流进步。