Spring rsocket security with Webflux security

Question

My Application is a Spring Webflux application with Spring boot version 2.6.6. Since, I have a chat and notification requirement for the logged in user, trying to use RSocket over websocket for notification & messaging along with Webflux for web based application.

Using Spring security for my web application with the config below and it is working. Now, not sure if I will be able to use the same security for RSocket as RSocket over websocket will be established when the user is logged in.

My Webflux security,

/**
 * 
 */
package com.TestApp.service.admin.spring.security;

import static java.util.stream.Collectors.toList;
import static org.springframework.security.config.Customizer.withDefaults;

import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.security.reactive.PathRequest;
import org.springframework.context.annotation.Bean;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.messaging.rsocket.RSocketStrategies;
import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
import org.springframework.security.authentication.ReactiveAuthenticationManager;
import org.springframework.security.config.annotation.rsocket.EnableRSocketSecurity;
import org.springframework.security.config.annotation.rsocket.RSocketSecurity;
import org.springframework.security.config.annotation.web.reactive.EnableWebFluxSecurity;
import org.springframework.security.config.web.server.SecurityWebFiltersOrder;
import org.springframework.security.config.web.server.ServerHttpSecurity;
import org.springframework.security.core.Authentication;
import org.springframework.security.core.context.ReactiveSecurityContextHolder;
import org.springframework.security.core.context.SecurityContext;
import org.springframework.security.messaging.handler.invocation.reactive.AuthenticationPrincipalArgumentResolver;
import org.springframework.security.rsocket.core.PayloadSocketAcceptorInterceptor;
import org.springframework.security.web.server.SecurityWebFilterChain;
import org.springframework.security.web.server.authentication.ServerAuthenticationFailureHandler;
import org.springframework.security.web.server.authentication.ServerAuthenticationSuccessHandler;
import org.springframework.security.web.server.authentication.logout.LogoutWebFilter;
import org.springframework.security.web.server.authentication.logout.RedirectServerLogoutSuccessHandler;
import org.springframework.security.web.server.authentication.logout.ServerLogoutHandler;
import org.springframework.security.web.server.authentication.logout.ServerLogoutSuccessHandler;
import org.springframework.security.web.server.authorization.HttpStatusServerAccessDeniedHandler;
import org.springframework.security.web.server.context.ServerSecurityContextRepository;
import org.springframework.security.web.server.csrf.CookieServerCsrfTokenRepository;
import org.springframework.security.web.server.util.matcher.AndServerWebExchangeMatcher;
import org.springframework.security.web.server.util.matcher.OrServerWebExchangeMatcher;
import org.springframework.security.web.server.util.matcher.PathPatternParserServerWebExchangeMatcher;
import org.springframework.security.web.server.util.matcher.ServerWebExchangeMatcher;
import org.springframework.security.web.server.util.matcher.ServerWebExchangeMatcher.MatchResult;
import org.springframework.security.web.server.util.matcher.ServerWebExchangeMatchers;
import org.springframework.web.server.WebSession;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.testapp.service.admin.spring.TestAppProperties;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;


@EnableWebFluxSecurity
public class AdminSecurityConfig {

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

    private static final String[] DEFAULT_FILTER_MAPPING = new String[] { "/**" };
    private static final String authenticateHeaderValue = "TestApp";
    private static final String unauthorizedJsonBody = "{\"message\": \"You are not authorized\"}";

    @Autowired
    private TestAppProperties testAppProps;

    @Bean
    public SecurityWebFilterChain securitygWebFilterChain(final ServerHttpSecurity http,
            final ReactiveAuthenticationManager authManager,
            final ServerSecurityContextRepository securityContextRepository,
            final TestAppAuthenticationFailureHandler failureHandler,
            final ObjectProvider<TestAppLogoutHandler> availableLogoutHandlers) {

        http.securityContextRepository(securityContextRepository);

        return http.authorizeExchange().matchers(PathRequest.toStaticResources().atCommonLocations()).permitAll()
                .pathMatchers(TestAppProps.getSecurity().getIgnorePatterns()).permitAll()
                .anyExchange().authenticated().and().formLogin().loginPage(TestAppProps.getSecurity().getLoginPath())
                .authenticationSuccessHandler(authSuccessHandler()).and().exceptionHandling()
                .authenticationEntryPoint((exchange, exception) -> Mono.error(exception))
                .accessDeniedHandler(new HttpStatusServerAccessDeniedHandler(HttpStatus.UNAUTHORIZED)).and().build();
    }

    @Bean
    public ServerAuthenticationSuccessHandler authSuccessHandler() {
        return new TestAppAuthSuccessHandler("/");
    }

    @Bean
    public ServerLogoutSuccessHandler logoutSuccessHandler(String uri) {
        RedirectServerLogoutSuccessHandler successHandler = new RedirectServerLogoutSuccessHandler();
        successHandler.setLogoutSuccessUrl(URI.create(uri));
        return successHandler;
    }

    @Bean(name = "failure-handler-bean")
    public TestAppAuthenticationFailureHandler defaultFailureHandler() {
        try {
            new ObjectMapper().reader().readTree(unauthorizedJsonBody);
        } catch (final IOException e) {
            throw new IllegalArgumentException("'unauthorizedJsonBody' property is not valid JSON.", e);
        }

        return new TestAppAdminAuthFailureHandler(authenticateHeaderValue, unauthorizedJsonBody);
    }

    @Bean
    public AuthenticatedPrinciplaProvider TestAppSecurityPrincipalProvider() {
        return new TestAppSecurityContextPrincipleProvider();
    }


}

public class TestAppSecurityContextPrincipleProvider implements AuthenticatedPrinciplaProvider {
  
  @Override
    public Mono<WhskrUserDetails> retrieveUser() {
        return principalMono.flatMap(principal -> {
            if (principal instanceof UsernamePasswordAuthenticationToken) {
                final TestAppUserDetails user = (TestAppUserDetails) ((UsernamePasswordAuthenticationToken) principal)
                        .getPrincipal();
                LOGGER.debug("User principal found for ID {} Org {} ", user.getUserId(), user.getOrgId());

                return Mono.just(user);
            }

            return Mono.error(() -> new IllegalArgumentException(NO_USER_AUTH_ERROR));
        })
    }
}

This is working as expected. Have a login page and user gets redirected to the home page after the successful login.

Now, I am adding RSocket over websocket for messaging and notification for the logged in user.

implementation 'org.springframework.boot:spring-boot-starter-webflux'
implementation 'org.springframework.boot:spring-boot-starter-security'
implementation 'org.springframework.security:spring-security-messaging'
implementation 'org.springframework.security:spring-security-rsocket'
implementation 'org.springframework.boot:spring-boot-starter-rsocket'
 

RSocketSecurityConfig,

@EnableWebFluxSecurity
public class AdminRSocketSecurityConfig {

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

    private static final String[] DEFAULT_FILTER_MAPPING = new String[] { "/**" };
    private static final String authenticateHeaderValue = "TestApp";
    private static final String unauthorizedJsonBody = "{\"message\": \"You are not authorized\"}";

    @Autowired
    private TestAppProperties TestAppProps;
    
    @Autowired
    private AuthenticatedPrinciplaProvider secContext;

    static final String RSOCKET_CONVERTER_BEAN_NAME = "RSocketAuthConverter";

    private static final String HEADERS = "headers";
    private static final MimeType COMPOSITE_METADATA_MIME_TYPE = MimeTypeUtils
            .parseMimeType(WellKnownMimeType.MESSAGE_RSOCKET_COMPOSITE_METADATA.getString());
    private static final MimeType APPLICATION_JSON_MIME_TYPE = MimeTypeUtils
            .parseMimeType(WellKnownMimeType.APPLICATION_JSON.getString());

    
    @Bean
    public RSocketStrategies rsocketStrategies() {
        return RSocketStrategies.builder()
            .encoders(encoders -> encoders.add(new Jackson2CborEncoder()))
            .decoders(decoders -> decoders.add(new Jackson2CborDecoder()))
            .routeMatcher(new PathPatternRouteMatcher())
            .build();
    }
    
    @Bean
    public RSocketMessageHandler messageHandler(RSocketStrategies strategies) {
        RSocketMessageHandler handler = new RSocketMessageHandler();
        HandlerMethodArgumentResolver resolver = new AuthenticationPrincipalArgumentResolver();
        handler.getArgumentResolverConfigurer().addCustomResolver(resolver);
        handler.setRSocketStrategies(strategies);
        return handler;
    }
    
    @Bean
    public PayloadSocketAcceptorInterceptor authorization(final ReactiveAuthenticationManager authManager,
            final RSocketSecurity security) {
        security.authorizePayload(authorize -> authorize.setup().authenticated()).authenticationManager(authManager);
        return security.build();
    }


}

RSocketController,

@Controller
public class RSocketController {

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

    private static final Map<Integer, Map<Integer, RSocketRequester>> CLIENT_REQUESTER_MAP = new HashMap<>();

    static final String SERVER = "Server";
    static final String RESPONSE = "Response";
    static final String STREAM = "Stream";
    static final String CHANNEL = "Channel";

    @Autowired
    private AuthenticatedPrinciplaProvider secContext;

    @ConnectMapping
    // void onConnect(RSocketRequester rSocketRequester, @Payload Integer userId) {
    void onConnect(RSocketRequester rSocketRequester) {
        secContext.retrieveUser().flatMap(usr -> {
            LOGGER.info("Client connect request for userId {} ", usr.getUserId());
            rSocketRequester.rsocket().onClose().doFirst(() -> {
                CLIENT_REQUESTER_MAP.put(usr.getUserId(), rSocketRequester);
            }).doOnError(error -> {
                LOGGER.info("Client connect request for userId {} ", usr.getUserId());
            }).doFinally(consumer -> {
                LOGGER.info("Removing here for userId {} ", usr.getUserId());
                if (CLIENT_REQUESTER_MAP.get(usr.getBranchId()) != null) {
                    CLIENT_REQUESTER_MAP.remove(usr.getUserId(), rSocketRequester);
                }
            }).subscribe();

            return Mono.empty();
        }).subscribe();

    }
}

From the RSocket over WebSocket client, the call is not going to the controller as auth is failing.

But, When I set "authorize.setup().permitAll()" in my RSocketSecurityConfig authorization(), the call goes to the controller, but the retrieveUser() fails.

I am not sure, How can I use the same security which is being used for my web based application for RSocket security as well?

So, When user is not logged in to my web app, the rsocket over websocket should fail and it should work only when the user is logged in. The RSocket initial call is happening once the user is logged in.