ActiveMq-拦截创建消息队列

Posted 2020-08-29 嫣然居

ActiveMQ拦截客户端创建/接收消息队列

1.创建插件

public class AuthPlugin implements BrokerPlugin{
    private String mqName;//本MQ服务器名称
    private JdbcTemplate jdbcTemplate;//数据库操作类
    
    public AuthPlugin(JdbcTemplate jdbcTemplate,String mqName) {
        this.jdbcTemplate=jdbcTemplate;
        this.mqName=mqName;
    }
    
    @Override
    public Broker installPlugin(Broker broker) throws Exception {
        return new AuthBroker(broker,jdbcTemplate,mqName);
    }
}

2.修改apache-activemq\conf\activemq.xml

<!--broker节点下-->
<plugins>
    <bean xmlns="http://www.springframework.org/schema/beans" id="ehlPlugin" class="com.ehl.plugin.AuthPlugin">
        <constructor-arg index="0">
            <ref bean="jdbcTemplate"/>
        </constructor-arg>
        <constructor-arg index="1" value="MQName"/><!--本消息队列的名称-->
    </bean>
</plugins>

3.创建插件类

public class AuthBroker extends AbstractAuthenticationBroker{
    private static Log log = LogFactory.getLog(AuthBroker.class);
    
    private static final String ACTIVEMQ_ADVISORY_PRODUCER_QUEUE="ActiveMQ.Advisory.Producer.Queue.";//消息生产者前缀
    private static final String ACTIVEMQ_ADVISORY_CONSUMER_QUEUE="ActiveMQ.Advisory.Consumer.Queue.";//消息消费者前缀
    private JdbcTemplate jdbcTemplate;//数据库操作
    private String mqName;//MQ服务器名称

    public AuthBroker(Broker next,JdbcTemplate jdbcTemplate,String mqName) {
        super(next);
        this.jdbcTemplate=jdbcTemplate;
        this.mqName=mqName;
    }
        /**
     * 连接拦截器
     */
    @Override
    public void addConnection(ConnectionContext context, ConnectionInfo info) throws Exception {
        log.info("用户["+info.getUserName()+"]请求连接["+mqName+"]!");
        SecurityContext securityContext = context.getSecurityContext();
        if (securityContext == null) {
            securityContext = authenticate(info.getUserName(), info.getPassword(), null);
            context.setSecurityContext(securityContext);
            securityContexts.add(securityContext);
        }
        
        try {
            super.addConnection(context, info);
        } catch (Exception e) {
            securityContexts.remove(securityContext);
            context.setSecurityContext(null);
            throw e;
        }
    }
        /**
     * 认证
     * <p>Title: authenticate</p>
     */
    @Override
    public SecurityContext authenticate(String username, String password, X509Certificate[] peerCertificates) throws SecurityException {
        SecurityContext securityContext = null;
        Com com=getCom(username,password);
        //验证用户信息
        if(com!=null&&com.getId()!=null){
             securityContext = new SecurityContext(username) {
                    @Override
                    public Set<Principal> getPrincipals() {
                        Set<Principal> groups = new HashSet<Principal>();
                        groups.add(new GroupPrincipal("users"));//默认加入了users的组
                        return groups;
                    }
                };
//                log.info("用户："+username+"验证成功!");
        }else{
            log.error("用户:"+username+"验证失败!");
            throw new SecurityException("验证失败");
        }
        return securityContext;
    }
        /**
     * 添加一个目标
     * <p>Title: addDestination</p>
     * @see org.apache.activemq.broker.BrokerFilter#addDestination(org.apache.activemq.broker.ConnectionContext, org.apache.activemq.command.ActiveMQDestination, boolean)
     */
    @Override
    public Destination addDestination(ConnectionContext context, ActiveMQDestination destination, boolean createIfTemporary) throws Exception {
        boolean destStats = destination.getPhysicalName().regionMatches(true, 0, ACTIVEMQ_ADVISORY_PRODUCER_QUEUE, 0,ACTIVEMQ_ADVISORY_PRODUCER_QUEUE.length());
        //发送消息者
        if(destStats){
            if(context.getSecurityContext()!=null){
                //判断不是默认用户
                if(!context.getSecurityContext().getUserName().equals(SecurityContext.BROKER_SECURITY_CONTEXT.getUserName())){
                    String queuesName=destination.getPhysicalName().replace(ACTIVEMQ_ADVISORY_PRODUCER_QUEUE, "");//得到消息队列名
                    if(powers.containsKey(context.getSecurityContext().getUserName())){//判断该用户是否有权限
                        Map<String,ViewProjectMqQueuesCom> map=powers.get(context.getSecurityContext().getUserName());
                        if(map!=null&&map.containsKey(queuesName)){//判断是否有发送的权限
                            if(map.get(queuesName).getBindId()!=null&&map.get(queuesName).getComQueuesType()!=null){
                                if(map.get(queuesName).getComQueuesType().intValue()==QueuesComType.BOTH.getValue().intValue()||map.get(queuesName).getComQueuesType().intValue()==QueuesComType.SEND.getValue().intValue()){
                                    return super.addDestination(context, destination, createIfTemporary);
                                }
                            }
                        }
                        throw new Exception("["+mqName+"-"+context.getUserName()+"]对消息队列["+queuesName+"]没有发送消息的权限");
                    }else{
                        throw new Exception("请登录后再操作!");
                    }
                }
            }
        }else{
            boolean consumerStats = destination.getPhysicalName().regionMatches(true, 0, ACTIVEMQ_ADVISORY_CONSUMER_QUEUE, 0,ACTIVEMQ_ADVISORY_CONSUMER_QUEUE.length());
            //消息接收者
            if(consumerStats){
                if(context.getSecurityContext()!=null){
                    //判断不是默认用户
                    if(!context.getSecurityContext().getUserName().equals(SecurityContext.BROKER_SECURITY_CONTEXT.getUserName())){
                        String queuesName=destination.getPhysicalName().replace(ACTIVEMQ_ADVISORY_CONSUMER_QUEUE, "");//得到消息队列名称
                        if(powers.containsKey(context.getSecurityContext().getUserName())){//判断用户是否有对应的权限
                            Map<String,ViewProjectMqQueuesCom> map=powers.get(context.getSecurityContext().getUserName());
                            if(map!=null&&map.containsKey(queuesName)){
                                if(map.get(queuesName).getBindId()!=null&&map.get(queuesName).getComQueuesType()!=null){
                                    if(map.get(queuesName).getComQueuesType().intValue()==QueuesComType.BOTH.getValue().intValue()||map.get(queuesName).getComQueuesType().intValue()==QueuesComType.RECEIVE.getValue().intValue()){
                                        return super.addDestination(context, destination, createIfTemporary);
                                    }
                                }
                            }
                            throw new Exception("["+mqName+"-"+context.getUserName()+"]对消息队列["+queuesName+"]没有获取消息的权限");
                        }else{
                            throw new Exception("请登录后再操作!");
                        }
                    }
                }
            }
        }
        return super.addDestination(context, destination, createIfTemporary);
    }
        /**
     * 监控发送消息
     * <p>Title: send</p>
     * @see org.apache.activemq.broker.BrokerFilter#send(org.apache.activemq.broker.ProducerBrokerExchange, org.apache.activemq.command.Message)
     */
    @Override
    public void send(ProducerBrokerExchange producerExchange, Message messageSend) throws Exception {
        String userName=producerExchange.getConnectionContext().getUserName();
        ActiveMQDestination msgDest = messageSend.getDestination();
        String physicalName = msgDest.getPhysicalName();
    }
/**
     * 监控消息接收者
     * <p>Title: acknowledge</p>
     * @see org.apache.activemq.broker.BrokerFilter#acknowledge(org.apache.activemq.broker.ConsumerBrokerExchange, org.apache.activemq.command.MessageAck)
     */
    @Override
    public void acknowledge(ConsumerBrokerExchange consumerExchange, MessageAck ack) throws Exception {
        String userName=consumerExchange.getConnectionContext().getUserName();
        String queues=ack.getDestination().getPhysicalName();
    }
}