带有rabbitmq的Spring websocket - 在订阅级别添加安全性

Question

在我的spring-boot 应用程序中，我有spring-security 和spring-websocket。下面是我的 websocket 配置。

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig extends WebSocketMessageBrokerConfigurationSupport
        implements WebSocketMessageBrokerConfigurer {

    @Value( "${rabbitmq.host}" )
    private String rabbitmqHost;
    @Value( "${rabbitmq.stomp.port}" )
    private int rabbitmqStompPort;
    @Value( "${rabbitmq.username}" )
    private String rabbitmqUserName;
    @Value( "${rabbitmq.password}" )
    private String rabbitmqPassword;

    @Override
    public void configureMessageBroker( MessageBrokerRegistry registry )
    {
        registry.enableStompBrokerRelay("/topic", "/queue").setRelayHost(rabbitmqHost).setRelayPort(rabbitmqStompPort)
                .setSystemLogin(rabbitmqUserName).setSystemPasscode(rabbitmqPassword);
        registry.setApplicationDestinationPrefixes("/app");
    }

    @Override
    public void registerStompEndpoints( StompEndpointRegistry stompEndpointRegistry )
    {
        stompEndpointRegistry.addEndpoint("/ws")
                .setAllowedOrigins("*")
                .withSockJS();
    }
}

还有，

public class CustomSubProtocolWebSocketHandler extends SubProtocolWebSocketHandler {

    private static final Logger LOGGER = LoggerFactory.getLogger(CustomSubProtocolWebSocketHandler.class);

    @Autowired
    private UserCommons userCommons;

    CustomSubProtocolWebSocketHandler(MessageChannel clientInboundChannel,
                                      SubscribableChannel clientOutboundChannel) {
        super(clientInboundChannel, clientOutboundChannel);
    }

    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        LOGGER.info("************************************************************************************************************************New webSocket connection was established: {}", session);
        String token = session.getUri().getQuery().replace("token=", "");
        try
        {
            String user = Jwts.parser().setSigningKey(TokenConstant.SECRET)
                    .parseClaimsJws(token.replace(TokenConstant.TOKEN_PREFIX, "")).getBody().getSubject();
            Optional<UserModel> userModelOptional = userCommons.getUserByEmail(user);
            if( !userModelOptional.isPresent() )
            {
                LOGGER.error(
                        "************************************************************************************************************************Invalid token is passed with web socket request");
                throw new DataException(GeneralConstants.EXCEPTION, "Invalid user", HttpStatus.BAD_REQUEST);
            }
        }
        catch( Exception e )
        {
            LOGGER.error(GeneralConstants.ERROR, e);
        }
        super.afterConnectionEstablished(session);
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) throws Exception {
        LOGGER.error("************************************************************************************************************************webSocket connection was closed");
        LOGGER.error("Reason for closure {} Session: {} ", closeStatus.getReason(),session.getId() );
        super.afterConnectionClosed(session, closeStatus);
    }

    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {

        LOGGER.error("************************************************************************************************************************Connection closed unexpectedly");
        LOGGER.error(GeneralConstants.ERROR, exception);
        super.handleTransportError(session, exception);
    }
}

为了在建立连接时添加安全层，我接受了连接 U​​RL 中的令牌。所以客户端应用程序将连接到/ws?token=*****。

但是要将消息发送给特定用户，我正在使用 user_id 构建订阅 URL。例如，如果登录的用户 id 是 23，客户端将订阅/topic/noti.23，然后从服务器端发送消息到/topic/noti.23。

 public void sendMessagesToTheDestination( WebSocketNotificationResponseBean webSocketNotificationResponseBean,
                List<String> paths )
        {
            try
            {
                for( String path : paths )
                {
                    LOGGER.info("Sending message to path: {}", path);
                    messagingTemplate.convertAndSend(path, webSocketNotificationResponseBean);
                    LOGGER.info("Sent message to path: {}", path);
                }
            }
            catch( Exception e )
            {
                LOGGER.error("Error while sending web socket notification {}", e);
            }
        }
}

其中path 是/topic/noti.<user_id>。

上面的实现是有效的。

现在的问题是，任何拥有有效令牌的用户都可以连接到 websocket，之后可以从浏览器控制台手动订阅任何 URL。比如user_id为23的用户，可以到浏览器控制台添加sockjs CDN，订阅/topic/noti.56，开始接收user_id为56的用户的消息。

这里如何添加安全层？

我尝试使用convertAndSendToUser，但不了解会话部分关于服务器如何理解会话以及我应该如何从客户端订阅。

谢谢

Answer 1

您不需要动态构建目标。您可以订阅像“/user/queue/wishes”这样的目的地，并且仍然可以发送私人消息。

String queueName = "/user/" + username  + "/queue/wishes";
simpMessagingTemplate.convertAndSend(queueName, message);