带有Spring Security和Rabbitmq的OAuth2授权 [英] OAuth2 authorization with Spring Security and Rabbitmq

问题描述

我们目前有许多与REST端点和RabbitMQ队列进行通信的Spring微服务.我们刚刚在所有服务上实现了OAuth2安全性，并且REST端点得到了适当的保护.

We currently have a number of Spring microservices that are communicating with REST endpoints and RabbitMQ queues. We have just implemented OAuth2 security on all of the services, and the REST endpoints are appropriately secured.

我们有一个编写的库可以创建RabbitTemplate和AmqpAdmin Bean，因此不必在每个服务中都完成样板代码.我们正在使用特定用户(普通用户和管理员)连接到Spring中的RabbitMQ服务器.我们不想以个人用户身份连接到RabbitMQ服务器.

We have a library that we wrote which creates the RabbitTemplate and AmqpAdmin beans so that the boilerplate code doesn't have to be done in every service. We are connecting to the RabbitMQ server in Spring with a specific user for regular clients, and another for admins. We don't want to connect to the RabbitMQ server as the individual user.

如果我们在Rabbit消息头中传递访问令牌，是否可以将RabbitTemplate配置为在处理消息之前检查令牌?是否可以/应该在模板的AfterReceive/BeforePublish处理器中全局执行此操作?还是需要在每个侦听器方法中单独检查此内容?

Is it possible, if we pass the access token in the rabbit message header, to configure the RabbitTemplate to check the token before the message gets handled? Is this something that can/should be done in the AfterReceive/BeforePublish processors globally for the template? Or will this need to be checked individually in each listener method?

谢谢

推荐答案

我能够通过创建自定义MessageListenerContainerFactory和MessageListenerContainer来制定解决方案.

I was able to work out a solution by creating a custom MessageListenerContainerFactory and MessageListenerContainer.

CustomMessageListenerContainerFactory.java:

CustomMessageListenerContainerFactory.java:

public class CustomMessageListenerContainerFactory extends AbstractRabbitListenerContainerFactory {

    DefaultTokenServices tokenServices;

    public CustomMessageListenerContainerFactory(DefaultTokenServices tokenServices) {
        this.tokenServices = tokenServices;
    }

    /**
     * Create an empty container instance.
     *
     * @return the new container instance.
     */
    @Override
    protected CustomMessageListenerContainer createContainerInstance() {
        return new CustomMessageListenerContainer(tokenServices);
    }
}

CustomMessageListenerContainer.java:

CustomMessageListenerContainer.java:

public class CustomMessageListenerContainer extends SimpleMessageListenerContainer {
    private final static String errorMessage = "No valid credentials found in request: {}";
    private final static String handlingMessage = "Handling queue: {}";
    private final static String receivedMessage = "Received Message: {}";
    private final static Logger logger = LoggerFactory.getLogger(CustomMessageListenerContainer.class);


    private DefaultTokenServices tokenServices;

    /**
     * Constructor
     *
     * @param tokenServices The instance of DefaultTokenServices used to decrypt the access token.
     */
    public CustomMessageListenerContainer(DefaultTokenServices tokenServices) {
        this.tokenServices = tokenServices;
    }

    /**
     * This method checks to see if there is a valid authorization
     *
     * @param channel   The AMQP channel on which the message was published.
     * @param messageIn The incoming message.
     * @throws Exception Throws an exception when there are no valid credentials in the message.
     */
    @Override
    protected void executeListener(Channel channel, Message messageIn) throws Exception {
        logger.info(handlingMessage, (Object[]) getQueueNames());
        logger.info(receivedMessage, BeanUtils.beanProperties(messageIn));
        if (messageIn.getMessageProperties().getHeaders().keySet().stream().anyMatch(t -> Objects.equals(t.toLowerCase(), "authorization"))) {
            String accessKey = messageIn.getMessageProperties()
                    .getHeaders()
                    .keySet()
                    .stream()
                    .filter(t -> Objects.equals(t.toLowerCase(), "authorization"))
                    .findFirst()
                    .get();
            OAuth2Authentication auth = tokenServices.loadAuthentication(messageIn.getMessageProperties().getHeaders().get(accessKey).toString());
            // If the token is expired, there will be no auth.
            if (auth != null) {
                SecurityContextHolder.getContext().setAuthentication(auth);
                super.executeListener(channel, messageIn);
                return;
            }
        }
        rejectMessage(channel, messageIn);
    }

    private void rejectMessage(Channel channel, Message messageIn) throws Exception {
        logger.info(errorMessage, (Object[]) getQueueNames());
        String localMessage = errorMessage.replace("{}", String.join(", ", getQueueNames()));
        if (messageIn.getMessageProperties().getReplyTo() != null) {
            channel.basicPublish("",
                    messageIn.getMessageProperties().getReplyTo(),
                    new AMQP.BasicProperties.Builder()
                            .contentType("application/json")
                            .correlationId(messageIn.getMessageProperties().getCorrelationId())
                            .build(),
                    "{\"errorMessage\":\"".concat(localMessage).concat("\"}").getBytes());
        }
        throw new AmqpRejectAndDontRequeueException(localMessage);
    }
}

CustomRabbitListenerConfigurer.java:

CustomRabbitListenerConfigurer.java:

...
@Override
    public void configureRabbitListeners(final RabbitListenerEndpointRegistrar registrar) {
        registrar.setMessageHandlerMethodFactory(messageHandlerMethodFactory());
        CustomMessageListenerContainerfactory factory = new CustomMessageListenerContainerfactory(tokenServices);
        ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
        factory.setConnectionFactory(connectionFactory);
        registrar.setContainerFactory(factory);
    }
...

这篇关于带有Spring Security和Rabbitmq的OAuth2授权的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！