为什么 SimpUserRegistry 在 EC2 实例上无法正常工作

Question

我正在使用 SimpUserRegistry 上网 user-count（使用 getUserCount()）。它在我的本地机器上运行良好，但在只有弹性 IP 且没有负载均衡器的 AWS EC2 实例（尝试使用 Amazon Linux 和 Ubuntu）上运行良好。

EC2 上的问题是某些用户在连接时从未添加到注册表中，因此我得到了错误的结果。

我有 session 个监听器，用于 SessionConnectedEvent 和 SessionDisconnectEvent，我在其中使用 SimpUserRegistry（自动装配）来获取用户存在。如果重要的话，我也是 SimpUserRegistry 是一个消息控制器。

以下是 websocket 消息代理配置：

@Order(Ordered.HIGHEST_PRECEDENCE + 99)
@Configuration
@EnableWebSocketMessageBroker
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class WebSocketMessageBrokerConfig extends AbstractWebSocketMessageBrokerConfigurer {

    @NonNull
    private SecurityChannelInterceptor securityChannelInterceptor;

    @Override
    public void configureMessageBroker(MessageBrokerRegistry config) {
        ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
        threadPoolTaskScheduler.setPoolSize(1);
        threadPoolTaskScheduler.setThreadGroupName("cb-heartbeat-");
        threadPoolTaskScheduler.initialize();

        config.enableSimpleBroker("/queue/", "/topic/")
                .setTaskScheduler(threadPoolTaskScheduler)
                .setHeartbeatValue(new long[] {1000, 1000});

        config.setApplicationDestinationPrefixes("/app");
    }

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

    @Override
    public void configureClientInboundChannel(ChannelRegistration registration) {
        registration.interceptors(securityChannelInterceptor);
    }
}

下面是上面配置中使用的通道拦截器class:

@Slf4j
@Component
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class SecurityChannelInterceptor extends ChannelInterceptorAdapter {

    @NonNull
    private SecurityService securityService;


    @Value("${app.auth.token.header}")
    private String authTokenHeader;



    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {
        StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
        StompCommand command = accessor.getCommand();

        if (StompCommand.CONNECT.equals(command)) {
            List<String> authTokenList = accessor.getNativeHeader(authTokenHeader);
            if (authTokenList == null || authTokenList.isEmpty()) {
                throw new AuthenticationFailureException("STOMP " + command + " missing " + this.authTokenHeader + " header!");
            }
            String accessToken = authTokenList.get(0);
            AppAuth authentication = securityService.authenticate(accessToken);
            log.info("STOMP {} authenticated. Authentication Token = {}", command, authentication);
            accessor.setUser(authentication);
            SecurityContextHolder.getContext().setAuthentication(authentication);

            Principal principal = accessor.getUser();
            if (principal == null) {
                throw new RuntimeException("StompHeaderAccessor did not set the authenticated User for " + authentication);
            }
        }

        return message;
    }

}

我还有以下计划任务，它只是每两秒打印一次用户名：

@Component
@Slf4j
@AllArgsConstructor(onConstructor = @__(@Autowired))
public class UserRegistryLoggingTask {

    private SimpUserRegistry simpUserRegistry;

    @Scheduled(fixedRate = 2000)
    public void logUsersInUserRegistry() {
        Set<String> userNames = simpUserRegistry.getUsers().stream().map(u -> u.getName()).collect(Collectors.toSet());
        log.info("UserRegistry has {} users with IDs {}", userNames.size(), userNames);
    }
}

有些用户名即使在连接后也不会显示。

执行SecurityServiceclass-

@Service
@AllArgsConstructor(onConstructor = @__(@Autowired))
public class SecurityService {

    private UserRepository userRepository;
    private UserCredentialsRepository userCredentialsRepository;
    private JwtHelper jwtHelper;

    public User getUser() {
        AppAuth auth = (AppAuth) SecurityContextHolder.getContext().getAuthentication();
        User user = (User) auth.getUser();
        return user;
    }

    public AppAuth authenticate(String accessToken) {
        String username = jwtHelper.tryExtractSubject(accessToken);
        if (username == null) {
            throw new AuthenticationFailureException("Invalid access token!");
        }

        User user = userRepository.findByUsername(username);
        if (user == null) {
            throw new AuthenticationFailureException("Invalid access token!");
        }

        AppAuth authentication = new AppAuth(user);
        return authentication;
    }
}

更新

以下是浏览器上的 SockJS 日志示例 -

来自服务器的正确响应 user-name header:

>>> CONNECT
AccessToken:eyJhbGciOiJIUzUxMiJ9.eyJzdWIiOiJkb2cifQ.Wf8AO77LluHEfEv61TIvugEXxOqIXKjsJBO8QMQh-rF7tzf56lBkdpOruqc7UPf_Pmj6-dnHZ5raq2MnMpeG8Q
accept-version:1.1,1.0
heart-beat:10000,10000

<<< CONNECTED
version:1.1
heart-beat:1000,1000
user-name:5a590e411b96f841cc00027f

服务器响应不正确 user-name header:

>>> CONNECT
AccessToken:eyJhbGciOiJIUzUxMiJ9.eyJzdWIiOiJtb3VzZSJ9.wqX5X_CSdHD8_7PZPiSzftGCuPz1ClQU0-F9RHCqOIIkMLzI4rt31_EAaykc8VojK2KGS6DcycWfAdMr2edzYg
accept-version:1.1,1.0
heart-beat:10000,10000

<<< CONNECTED
version:1.1
heart-beat:1000,1000

我还验证了 SecurityChannelInterceptor 正在验证所有用户，即使 user-name 不在 CONNECTED 响应中也是如此。

更新

我在 heroku 上部署了该应用程序。问题也在那里发生。

更新

出现问题时，SessionConnectEvent中的user是SecurityChannelInterceptor设置的，但SessionConnectedEvent中的user是null。

更新

AppAuth class -

public class AppAuth implements Authentication {

    private final User user;
    private final Collection<GrantedAuthority> authorities;


    public AppAuth(User user) {
        this.user = user;
        this.authorities = Collections.singleton((GrantedAuthority) () -> "USER");
    }

    public User getUser() {
        return this.user;
    }

    @Override
    public String getName() {
        return user.getId();
    }

    @Override
    public Collection<? extends GrantedAuthority> getAuthorities() {
        return authorities;
    }

    @Override
    public Object getCredentials() {
        return null;
    }

    @Override
    public Object getDetails() {
        return null;
    }

    @Override
    public Object getPrincipal() {
        return new Principal() {
            @Override
            public String getName() {
                return user.getId();
            }
        };
    }

    @Override
    public boolean isAuthenticated() {
        return true;
    }

    @Override
    public void setAuthenticated(boolean isAuthenticated) throws IllegalArgumentException {

    }
}

Answer 1

以下几点可能会帮助您解决问题。

显然对于某些用户来说 authentication 没有被设置。您是否注意到没有设置 authentication 的任何模式？浏览 DefaultSimpUserRegistry and StompSubProtocolHandler 的源代码可以再次确认 authentication/principle 没有为某些用户设置。这就是 SimpUserRegistry 中缺少用户的原因。
经历这个 - 。它表明，如果 authentication/principle.

GrantedAuthorities

authentication

您或许可以检查 AppAuth 对象的创建，并查看是否为所有用户正确设置了 GrantedAuthorities。

希望这对解决您的问题有所帮助。

Answer 2

通过在 StompSubProtocolHandler 中添加一些记录器语句进行一些调试后，我能够跟踪问题。找到原因后，结论是通道拦截器不是对用户进行身份验证的正确位置。至少对于我的用例。

以下是 StompSubProtocolHandler -

中的一些代码片段

handleMessageFromClient 方法将用户添加到 stompAuthentications 地图并发布 SessionConnectEvent 事件 -

public void handleMessageFromClient(WebSocketSession session, WebSocketMessage<?> webSocketMessage, MessageChannel outputChannel) {
    //...
    SimpAttributesContextHolder.setAttributesFromMessage(message);
    boolean sent = outputChannel.send(message);

    if (sent) {
        if (isConnect) {
            Principal user = headerAccessor.getUser();
            if (user != null && user != session.getPrincipal()) {
                this.stompAuthentications.put(session.getId(), user);
            }
        }
        if (this.eventPublisher != null) {
            if (isConnect) {
                publishEvent(new SessionConnectEvent(this, message, getUser(session)));
            }
    //...

并且 handleMessageToClient 从 stompAuthentications 映射中检索用户并发布 SessionConnectedEvent -

public void handleMessageToClient(WebSocketSession session, Message<?> message) {
    //...
    SimpAttributes simpAttributes = new SimpAttributes(session.getId(), session.getAttributes());
    SimpAttributesContextHolder.setAttributes(simpAttributes);
    Principal user = getUser(session);
    publishEvent(new SessionConnectedEvent(this, (Message<byte[]>) message, user));
    //...

getUser以上方法使用的方法-

private Principal getUser(WebSocketSession session) {
    Principal user = this.stompAuthentications.get(session.getId());
    return user != null ? user : session.getPrincipal();
}

现在，当 handleMessageToClient 片段在 handleMessageFromClient 片段之前执行时会出现问题。在这种情况下，用户永远不会添加到 DefaultSimpUserRegistry，因为它只检查 SessionConnectedEvent.

下面是来自 DefaultSimpUserRegistry -

的事件侦听器片段

public void onApplicationEvent(ApplicationEvent event) {
    //...
    else if (event instanceof SessionConnectedEvent) {
        Principal user = subProtocolEvent.getUser();
        if (user == null) {
            return;
        }
    //...

解决方案

解决方案是扩展 DefaultHandshakeHandler and override determineUser method, which is based on . But, as I am using SockJS, this requires the client to send auth-token as a query parameter. And the reason for the query parameter requirement is discussed here。

为什么 SimpUserRegistry 在 EC2 实例上无法正常工作

Why is SimpUserRegistry not working properly on EC2 Instance

java

amazon-ec2

websocket

spring-websocket

spring-messaging