ActiveMQ是目前较为流行的一款开源消息服务器。最近在项目开发中，需要为ActiveMQ开发基于IP的验证和授权机制，因此，对ActiveMQ的安全机制进行了了解，以下将介绍ActiveMQ的安全机制使用及其源代码分析。
本文开发环境介绍：
操作系统：Windows XP
Java：jdk 1.6.0_12
maven：maven 3.0.4
ActiveMQ：ActiveMQ 5.6.0
ActiveMQ安全机制的介绍
安全机制一般包含验证（Authentication）和授权（Authorization）两部分。在ActiveMQ中，验证指通过访问者的用户名和密码实现用户身份的验证，授权指为消息目标（队列或主题）的读、写、管理指定具有相应权限的用户组，并为用户分配权限。ActiveMQ的安全机制基于插件实现。
ActiveMQ提供两种验证插件，分别是：
- Simple authentication plugin-in；
- JAAS（Java Authentication and Authorization Service）authentication plugin-in。
ActiveMQ提供一种授权插件：Authorization plugin-in。
ActiveMQ安全机制的使用
ActiveMQ的使用
可从ActiveMQ官网“http://activemq.apache.org/” 下载ActiveMQ的源代码包或二进制分发包。由于ActiveMQ使用Java开发，因此需要预先安装jdk，另外，由于ActiveMQ的开发使用了maven，因此，若下载的是源代码包，需要预先安装maven。解压源代码包，并在源代码包目录下执行“mvn install -Dmaven.test.skip=true ”完成编译、打包和安装，成功后，会在assembly\target下生成二进制分发包。若下载的是二进制分发包，解压即可。
ActiveMQ的二进制分发包目录如下所示：
进入bin文件，执行脚本，即可运行ActiveMQ。
Simple authentication plugin-in的使用
在activemq.xml中如下配置：1
2
3
4
5
6
7
8
9<plugins>
<simpleAuthenticationPlugin>
<users>
<authenticationUser username="system" password="password" groups="users,admins"/>
<authenticationUser username="user" password="password" groups="users"/>
<authenticationUser username="guest" password="password" groups="guests"/>
</users>
</simpleAuthenticationPlugin>
</plugins>
JAAS authentication plugin-in的使用
在activemq.xml中如下配置：1
2
3<plugins>
<jaasAuthenticationPlugin configuration="activemq-domain" />
</plugins>
创建login.config文件：1
2
3
4
5
6activemq-domain {
org.apache.activemq.jaas.PropertiesLoginModule required
debug=true
org.apache.activemq.jaas.properties.user="users.properties"
org.apache.activemq.jaas.properties.group="groups.properties";
};
创建users.properties和groups.properties文件，包含用户和用户组信息。
users.properties：1
2
3system=password
user=password
guest=password
groups.properties：1
2
3admins=system
users=system,user
guests=guest
Authorization plugin-in的使用
在activemq.xml中如下配置：1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18<plugins>
<authorizationPlugin>
<map>
<authorizationMap>
<authorizationEntries>
<authorizationEntry queue=">" read="admins" write="admins" admin="admins" />
<authorizationEntry queue="USERS.>" read="users" write="users" admin="users" />
<authorizationEntry queue="GUEST.>" read="guests" write="guests,users" admin="guests,users" />
<authorizationEntry queue="TEST.Q" read="guests" write="guests" />
<authorizationEntry topic=">" read="admins" write="admins" admin="admins" />
<authorizationEntry topic="USERS.>" read="users" write="users" admin="users" />
<authorizationEntry topic="GUEST.>" read="guests" write="guests,users" admin="guests,users" />
<authorizationEntry topic="ActiveMQ.Advisory.>" read="guests,users" write="guests,users" admin="guests,users"/>
</authorizationEntries>
</authorizationMap>
</map>
</authorizationPlugin>
</plugins>
ActiveMQ安全机制的源代码分析
ActiveMQ在其maven工程的activemq-core模块中实现安全机制。ActiveMQ原有安全机制均基于插件实现，实现思路如图所示。
其中，Broker接口是ActiveMQ的核心接口，ActiveMQ消息服务器对象即该接口的实现。接口BrokerPlugin通过installPlugin方法传入Broker对象，为其创建插件。BrokerFilter类也实现自Broker接口，其与Broker的关系，类似于Struts中Interceptor与Action的关系，多个BrokerFilter对象以及消息服务器Broker对象通过指向下一个对象的引用next构成链状结构，当创建连接、消息生产者、消息消费者时，先后执行BrokerFilter中的相应方法，直至执行消息服务器中的方法，而安全机制类即继承自BrokerFilter。
ActiveMQ原有安全机制的相关类均继承或实现自上述类或接口，安全机制的类包为activemq-core中的org.apache.activemq.security。
Simple authentication plugin-in的源代码分析
Simple authentication plugin-in主要包含两个基本类：SimpleAuthenticationPlugin（实现自BrokerPlugin）和SimpleAuthenticationBroker（继承自BrokerFilter）。
SimpleAuthenticationPlugin部分代码：1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19public class SimpleAuthenticationPlugin implements BrokerPlugin {
private Map<String, String> userPasswords;
private Map<String, Set<Principal>> userGroups;
private static final String DEFAULT_ANONYMOUS_USER = "anonymous";
private static final String DEFAULT_ANONYMOUS_GROUP = "anonymous";
private String anonymousUser = DEFAULT_ANONYMOUS_USER;
private String anonymousGroup = DEFAULT_ANONYMOUS_GROUP;
private boolean anonymousAccessAllowed = false;
//......
//安装插件时，根据activemq.xml中的配置，新建 SimpleAuthenticationBroker对象， 并返回该对象
public Broker installPlugin(Broker parent) {
SimpleAuthenticationBroker broker = new SimpleAuthenticationBroker(parent, userPasswords, userGroups);
broker.setAnonymousAccessAllowed(anonymousAccessAllowed);
broker.setAnonymousUser(anonymousUser);
broker.setAnonymousGroup(anonymousGroup);
return broker;
}
//......
}
SimpleAuthenticationBroker部分代码：1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58public class SimpleAuthenticationBroker extends BrokerFilter {
private boolean anonymousAccessAllowed = false;
private String anonymousUser;
private String anonymousGroup;
private final Map<String,String> userPasswords;
private final Map<String,Set<Principal>> userGroups;
private final CopyOnWriteArrayList<SecurityContext> securityContexts = new CopyOnWriteArrayList<SecurityContext>();
//......
//由于验证需要在创建连接时进行，因此重写BrokerFilter的addConnection方法
public void addConnection(ConnectionContext context, ConnectionInfo info) throws Exception {
SecurityContext s = context.getSecurityContext();
if (s == null) {
//若允许匿名访问，则不进行验证
// Check the username and password.
if (anonymousAccessAllowed && info.getUserName() == null && info.getPassword() == null) {
info.setUserName(anonymousUser);
s = new SecurityContext(info.getUserName()) {
public Set<Principal> getPrincipals() {
Set<Principal> groups = new HashSet<Principal>();
groups.add(new GroupPrincipal(anonymousGroup));
return groups;
}
};
//若不允许匿名访问，则验证连接的用户名和密码是否与配置文件中的一致，若不一致，则抛出安全异常
} else {
String pw = userPasswords.get(info.getUserName());
if (pw == null || !pw.equals(info.getPassword())) {
throw new SecurityException(
"User name [" + info.getUserName() + "] or password is invalid.");
}
final Set<Principal> groups = userGroups.get(info.getUserName());
s = new SecurityContext(info.getUserName()) {
public Set<Principal> getPrincipals() {
return groups;
}
};
}
context.setSecurityContext(s);
securityContexts.add(s);
}
//调用父对象的addConnection方法，即调用next引用的Broker对象的addConnection方法
try {
super.addConnection(context, info);
} catch (Exception e) {
securityContexts.remove(s);
context.setSecurityContext(null);
throw e;
}
}
//......
}
JAAS authentication plugin-in的源代码分析
JAAS authentication plugin-in主要包含两个基本类：JaasAuthenticationPlugin（实现自BrokerPlugin）JaasAuthenticationBroker（继承自BrokerFilter）。
JaasAuthenticationPlugin部分代码：1
2
3
4
5
6
7
8
9
10
11public class JaasAuthenticationPlugin implements BrokerPlugin {
protected String configuration = "activemq-domain";
//......
public Broker installPlugin(Broker broker) {
//读取配置文件， 初始化JAAS
initialiseJaas();
//创建JaasAuthenticationBroker对象并返回
return new JaasAuthenticationBroker(broker, configuration);
}
//......
}
JaasAuthenticationBroker部分代码：1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40public class JaasAuthenticationBroker extends BrokerFilter {
private final String jassConfiguration;
private final CopyOnWriteArrayList<SecurityContext> securityContexts = new CopyOnWriteArrayList<SecurityContext>();
//......
//由于验证需要在创建连接时进行，因此重写BrokerFilter的addConnection方法
public void addConnection(ConnectionContext context, ConnectionInfo info) throws Exception {
if (context.getSecurityContext() == null) {
// Set the TCCL since it seems JAAS needs it to find the login
// module classes.
ClassLoader original = Thread.currentThread().getContextClassLoader();
Thread.currentThread().setContextClassLoader(JaasAuthenticationBroker.class.getClassLoader());
try {
// Do the login.
try {
JassCredentialCallbackHandler callback = new JassCredentialCallbackHandler(info
.getUserName(), info.getPassword());
LoginContext lc = new LoginContext(jassConfiguration, callback);
lc.login();
Subject subject = lc.getSubject();
//基于JAAS判断用户名和密码是否正确
SecurityContext s = new JaasSecurityContext(info.getUserName(), subject);
context.setSecurityContext(s);
securityContexts.add(s);
} catch (Exception e) {
throw (SecurityException)new SecurityException("User name [" + info.getUserName() + "] or password is invalid.")
.initCause(e);
}
} finally {
Thread.currentThread().setContextClassLoader(original);
}
}
//调用父对象的addConnection方法，即调用next引用的Broker对象的addConnection方法
super.addConnection(context, info);
}
//......
}
Authorization plugin-in的源代码分析
Authorization plugin-in主要包含两个基本类：AuthorizationPlugin（实现自BrokerPlugin）AuthorizationBroker（继承自BrokerFilter）。
AuthorizationPlugin部分代码：1
2
3
4
5
6
7
8
9
10
11
12
13
14
15public class AuthorizationPlugin implements BrokerPlugin {
//AuthorizationMap对象存储activemq.xml中消息目标、读、写、管理用户组信息
private AuthorizationMap map;
//......
//创建 AuthorizationBroker 对象并返回
public Broker installPlugin(Broker broker) {
if (map == null) {
throw new IllegalArgumentException("You must configure a 'map' property");
}
return new AuthorizationBroker(broker, map);
}
//......
}
AuthorizationBroker部分代码：1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59public class AuthorizationBroker extends BrokerFilter implements SecurityAdminMBean {
private final AuthorizationMap authorizationMap;
//......
//由于需要授权是否可管理消息目标，因此重写BrokerFilter的 addDestinationInfo 方法
public void addDestinationInfo(ConnectionContext context, DestinationInfo info) throws Exception {
addDestination(context, info.getDestination(),true);
super.addDestinationInfo(context, info);
}
//由于需要授权是否可管理消息目标，因此重写BrokerFilter的 addDestination 方法
public Destination addDestination(ConnectionContext context, ActiveMQDestination destination,boolean create) throws Exception {
final SecurityContext securityContext = context.getSecurityContext();
if (securityContext == null) {
throw new SecurityException("User is not authenticated.");
}
Destination existing = this.getDestinationMap().get(destination);
if (existing != null) {
return super.addDestination(context, destination,create);
}
//从访问控制列表中查看是否具有授权
if (!securityContext.isBrokerContext()) {
Set<?> allowedACLs = null;
if (!destination.isTemporary()) {
allowedACLs = authorizationMap.getAdminACLs(destination);
} else {
allowedACLs = authorizationMap.getTempDestinationAdminACLs();
}
if (allowedACLs != null && !securityContext.isInOneOf(allowedACLs)) {
throw new SecurityException("User " + securityContext.getUserName() + " is not authorized to create: " + destination);
}
}
//调用next引用的addDestination方法
return super.addDestination(context, destination,create);
}
//由于需要授权是否可读消息，因此重写BrokerFilter的 addConsumer 方法，在该方法中，从访问控制列表中查看是否具有读授权，并调用next引用的addConsumer方法
public Subscription addConsumer(ConnectionContext context, ConsumerInfo info) throws Exception {
//......
return super.addConsumer(context, info);
}
//由于需要授权是否可写消息，因此重写BrokerFilter的 addProducer 方法，在该方法中，从访问控制列表中查看是否具有写授权，并调用next引用的 addProducer 方法
public void addProducer(ConnectionContext context, ProducerInfo info) throws Exception {
//......
super.addProducer(context, info);
}
//......
}
总结
ActiveMQ提供了便利的插件开发方式，并基于插件实现了包含验证和授权的安全机制。参考ActiveMQ的源代码，可以进行插件开发，实现个性化的安全机制，如基于IP的验证和授权。