Add RSocket Support

Fixes gh-7360

config/src/main/java/org/springframework/security/config/annotation/rsocket/EnableRSocketSecurity.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.security.config.annotation.rsocket;
+
+import org.springframework.context.annotation.Import;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Add this annotation to a {@code Configuration} class to have Spring Security
+ * {@link RSocketSecurity} support added.
+ *
+ * @author Rob Winch
+ * @since 5.2
+ * @see RSocketSecurity
+ */
+@Documented
+@Target(ElementType.TYPE)
+@Retention(RetentionPolicy.RUNTIME)
+@Import({ RSocketSecurityConfiguration.class })
+public @interface EnableRSocketSecurity { }

config/src/main/java/org/springframework/security/config/annotation/rsocket/RSocketSecurity.java

@@ -0,0 +1,313 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.security.config.annotation.rsocket;
+
+import org.springframework.beans.BeansException;
+import org.springframework.context.ApplicationContext;
+import org.springframework.core.ResolvableType;
+import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
+import org.springframework.security.authentication.ReactiveAuthenticationManager;
+import org.springframework.security.authorization.AuthenticatedReactiveAuthorizationManager;
+import org.springframework.security.authorization.AuthorityReactiveAuthorizationManager;
+import org.springframework.security.authorization.AuthorizationDecision;
+import org.springframework.security.authorization.ReactiveAuthorizationManager;
+import org.springframework.security.config.Customizer;
+import org.springframework.security.oauth2.jwt.ReactiveJwtDecoder;
+import org.springframework.security.oauth2.server.resource.authentication.JwtReactiveAuthenticationManager;
+import org.springframework.security.rsocket.interceptor.PayloadInterceptor;
+import org.springframework.security.rsocket.interceptor.PayloadSocketAcceptorInterceptor;
+import org.springframework.security.rsocket.interceptor.authentication.AnonymousPayloadInterceptor;
+import org.springframework.security.rsocket.interceptor.authentication.AuthenticationPayloadInterceptor;
+import org.springframework.security.rsocket.interceptor.authentication.BearerPayloadExchangeConverter;
+import org.springframework.security.rsocket.interceptor.authorization.AuthorizationPayloadInterceptor;
+import org.springframework.security.rsocket.interceptor.authorization.PayloadExchangeMatcherReactiveAuthorizationManager;
+import org.springframework.security.rsocket.util.PayloadExchangeAuthorizationContext;
+import org.springframework.security.rsocket.util.PayloadExchangeMatcher;
+import org.springframework.security.rsocket.util.PayloadExchangeMatcherEntry;
+import org.springframework.security.rsocket.util.PayloadExchangeMatchers;
+import org.springframework.security.rsocket.util.RoutePayloadExchangeMatcher;
+import reactor.core.publisher.Mono;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Allows configuring RSocket based security.
+ *
+ * A minimal example can be found below:
+ *
+ * <pre class="code">
+ * &#064;EnableRSocketSecurity
+ * public class SecurityConfig {
+ *     // @formatter:off
+ *     &#064;Bean
+ *     PayloadSocketAcceptorInterceptor rsocketInterceptor(RSocketSecurity rsocket) {
+ *         rsocket
+ *             .authorizePayload(authorize ->
+ *                 authorize
+ *                     .anyRequest().authenticated()
+ *             );
+ *         return rsocket.build();
+ *     }
+ *     // @formatter:on
+ *
+ *     // @formatter:off
+ *     &#064;Bean
+ *     public MapReactiveUserDetailsService userDetailsService() {
+ *          UserDetails user = User.withDefaultPasswordEncoder()
+ *               .username("user")
+ *               .password("password")
+ *               .roles("USER")
+ *               .build();
+ *          return new MapReactiveUserDetailsService(user);
+ *     }
+ *     // @formatter:on
+ * }
+ * </pre>
+ *
+ * A more advanced configuration can be seen below:
+ *
+ * <pre class="code">
+ * &#064;EnableRSocketSecurity
+ * public class SecurityConfig {
+ *     // @formatter:off
+ *     &#064;Bean
+ *     PayloadSocketAcceptorInterceptor rsocketInterceptor(RSocketSecurity rsocket) {
+ *         rsocket
+ *             .authorizePayload(authorize ->
+ *                 authorize
+ *                     // must have ROLE_SETUP to make connection
+ *                     .setup().hasRole("SETUP")
+ *                      // must have ROLE_ADMIN for routes starting with "admin."
+ *                     .route("admin.*").hasRole("ADMIN")
+ *                     // any other request must be authenticated for
+ *                     .anyRequest().authenticated()
+ *             );
+ *         return rsocket.build();
+ *     }
+ *     // @formatter:on
+ * }
+ * </pre>
+ * @author Rob Winch
+ * @since 5.2
+ */
+public class RSocketSecurity {
+
+	private BasicAuthenticationSpec basicAuthSpec;
+
+	private JwtSpec jwtSpec;
+
+	private AuthorizePayloadsSpec authorizePayload;
+
+	private ApplicationContext context;
+
+	private ReactiveAuthenticationManager authenticationManager;
+
+	public RSocketSecurity authenticationManager(ReactiveAuthenticationManager authenticationManager) {
+		this.authenticationManager = authenticationManager;
+		return this;
+	}
+
+	public RSocketSecurity basicAuthentication(Customizer<BasicAuthenticationSpec> basic) {
+		if (this.basicAuthSpec == null) {
+			this.basicAuthSpec = new BasicAuthenticationSpec();
+		}
+		basic.customize(this.basicAuthSpec);
+		return this;
+	}
+
+	public class BasicAuthenticationSpec {
+		private ReactiveAuthenticationManager authenticationManager;
+
+		public BasicAuthenticationSpec authenticationManager(ReactiveAuthenticationManager authenticationManager) {
+			this.authenticationManager = authenticationManager;
+			return this;
+		}
+
+		private ReactiveAuthenticationManager getAuthenticationManager() {
+			if (this.authenticationManager == null) {
+				return RSocketSecurity.this.authenticationManager;
+			}
+			return this.authenticationManager;
+		}
+
+		protected AuthenticationPayloadInterceptor build() {
+			ReactiveAuthenticationManager manager = getAuthenticationManager();
+			return new AuthenticationPayloadInterceptor(manager);
+		}
+
+		private BasicAuthenticationSpec() {}
+	}
+
+	public RSocketSecurity jwt(Customizer<JwtSpec> jwt) {
+		if (this.jwtSpec == null) {
+			this.jwtSpec = new JwtSpec();
+		}
+		jwt.customize(this.jwtSpec);
+		return this;
+	}
+
+	public class JwtSpec {
+		private ReactiveAuthenticationManager authenticationManager;
+
+		public JwtSpec authenticationManager(ReactiveAuthenticationManager authenticationManager) {
+			this.authenticationManager = authenticationManager;
+			return this;
+		}
+
+		private ReactiveAuthenticationManager getAuthenticationManager() {
+			if (this.authenticationManager != null) {
+				return this.authenticationManager;
+			}
+			ReactiveJwtDecoder jwtDecoder = getBeanOrNull(ReactiveJwtDecoder.class);
+			if (jwtDecoder != null) {
+				this.authenticationManager = new JwtReactiveAuthenticationManager(jwtDecoder);
+				return this.authenticationManager;
+			}
+			return RSocketSecurity.this.authenticationManager;
+		}
+
+		protected AuthenticationPayloadInterceptor build() {
+			ReactiveAuthenticationManager manager = getAuthenticationManager();
+			AuthenticationPayloadInterceptor result = new AuthenticationPayloadInterceptor(manager);
+			result.setAuthenticationConverter(new BearerPayloadExchangeConverter());
+			return result;
+		}
+
+		private JwtSpec() {}
+	}
+
+	public RSocketSecurity authorizePayload(Customizer<AuthorizePayloadsSpec> authorize) {
+		if (this.authorizePayload == null) {
+			this.authorizePayload = new AuthorizePayloadsSpec();
+		}
+		authorize.customize(this.authorizePayload);
+		return this;
+	}
+
+	public PayloadSocketAcceptorInterceptor build() {
+		PayloadSocketAcceptorInterceptor interceptor = new PayloadSocketAcceptorInterceptor(
+				payloadInterceptors());
+		RSocketMessageHandler handler = getBean(RSocketMessageHandler.class);
+		interceptor.setDefaultDataMimeType(handler.getDefaultDataMimeType());
+		interceptor.setDefaultMetadataMimeType(handler.getDefaultMetadataMimeType());
+		return interceptor;
+	}
+
+	private List<PayloadInterceptor> payloadInterceptors() {
+		List<PayloadInterceptor> payloadInterceptors = new ArrayList<>();
+
+		if (this.basicAuthSpec != null) {
+			payloadInterceptors.add(this.basicAuthSpec.build());
+		}
+		if (this.jwtSpec != null) {
+			payloadInterceptors.add(this.jwtSpec.build());
+		}
+		payloadInterceptors.add(new AnonymousPayloadInterceptor("anonymousUser"));
+
+		if (this.authorizePayload != null) {
+			payloadInterceptors.add(this.authorizePayload.build());
+		}
+		return payloadInterceptors;
+	}
+
+	public class AuthorizePayloadsSpec {
+
+		private PayloadExchangeMatcherReactiveAuthorizationManager.Builder authzBuilder =
+				PayloadExchangeMatcherReactiveAuthorizationManager.builder();
+
+		public Access setup() {
+			return matcher(PayloadExchangeMatchers.setup());
+		}
+
+		public Access anyRequest() {
+			return matcher(PayloadExchangeMatchers.anyExchange());
+		}
+
+		protected AuthorizationPayloadInterceptor build() {
+			return new AuthorizationPayloadInterceptor(this.authzBuilder.build());
+		}
+
+		public Access route(String pattern) {
+			RSocketMessageHandler handler = getBean(RSocketMessageHandler.class);
+			PayloadExchangeMatcher matcher = new RoutePayloadExchangeMatcher(
+					handler.getMetadataExtractor(),
+					handler.getRouteMatcher(),
+					pattern);
+			return matcher(matcher);
+		}
+
+		public Access matcher(PayloadExchangeMatcher matcher) {
+			return new Access(matcher);
+		}
+
+		public class Access {
+
+			private final PayloadExchangeMatcher matcher;
+
+			private Access(PayloadExchangeMatcher matcher) {
+				this.matcher = matcher;
+			}
+
+			public AuthorizePayloadsSpec authenticated() {
+				return access(AuthenticatedReactiveAuthorizationManager.authenticated());
+			}
+
+			public AuthorizePayloadsSpec hasRole(String role) {
+				return access(AuthorityReactiveAuthorizationManager.hasRole(role));
+			}
+
+			public AuthorizePayloadsSpec permitAll() {
+				return access((a, ctx) -> Mono
+						.just(new AuthorizationDecision(true)));
+			}
+
+			public AuthorizePayloadsSpec access(
+					ReactiveAuthorizationManager<PayloadExchangeAuthorizationContext> authorization) {
+				AuthorizePayloadsSpec.this.authzBuilder.add(new PayloadExchangeMatcherEntry<>(this.matcher, authorization));
+				return AuthorizePayloadsSpec.this;
+			}
+		}
+	}
+
+	private <T> T getBean(Class<T> beanClass) {
+		if (this.context == null) {
+			return null;
+		}
+		return this.context.getBean(beanClass);
+	}
+
+	private <T> T getBeanOrNull(Class<T> beanClass) {
+		return getBeanOrNull(ResolvableType.forClass(beanClass));
+	}
+
+	private <T> T getBeanOrNull(ResolvableType type) {
+		if (this.context == null) {
+			return null;
+		}
+		String[] names =  this.context.getBeanNamesForType(type);
+		if (names.length == 1) {
+			return (T) this.context.getBean(names[0]);
+		}
+		return null;
+	}
+
+	protected void setApplicationContext(ApplicationContext applicationContext)
+			throws BeansException {
+		this.context = applicationContext;
+	}
+}

config/src/main/java/org/springframework/security/config/annotation/rsocket/RSocketSecurityConfiguration.java

@@ -0,0 +1,84 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.security.config.annotation.rsocket;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Scope;
+import org.springframework.security.authentication.ReactiveAuthenticationManager;
+import org.springframework.security.authentication.UserDetailsRepositoryReactiveAuthenticationManager;
+import org.springframework.security.core.userdetails.ReactiveUserDetailsService;
+import org.springframework.security.crypto.password.PasswordEncoder;
+
+/**
+ * @author Rob Winch
+ * @since 5.2
+ */
+@Configuration(proxyBeanMethods = false)
+class RSocketSecurityConfiguration {
+
+	private static final String BEAN_NAME_PREFIX = "org.springframework.security.config.annotation.rsocket.RSocketSecurityConfiguration.";
+	private static final String RSOCKET_SECURITY_BEAN_NAME = BEAN_NAME_PREFIX + "rsocketSecurity";
+
+	private ReactiveAuthenticationManager authenticationManager;
+
+	private ReactiveUserDetailsService reactiveUserDetailsService;
+
+	private PasswordEncoder passwordEncoder;
+
+	@Autowired(required = false)
+	void setAuthenticationManager(
+			ReactiveAuthenticationManager authenticationManager) {
+		this.authenticationManager = authenticationManager;
+	}
+
+	@Autowired(required = false)
+	void setUserDetailsService(ReactiveUserDetailsService userDetailsService) {
+		this.reactiveUserDetailsService = userDetailsService;
+	}
+
+	@Autowired(required = false)
+	void setPasswordEncoder(PasswordEncoder passwordEncoder) {
+		this.passwordEncoder = passwordEncoder;
+	}
+
+	@Bean(name = RSOCKET_SECURITY_BEAN_NAME)
+	@Scope("prototype")
+	public RSocketSecurity rsocketSecurity(ApplicationContext context) {
+		RSocketSecurity security = new RSocketSecurity()
+			.authenticationManager(authenticationManager());
+		security.setApplicationContext(context);
+		return security;
+	}
+
+	private ReactiveAuthenticationManager authenticationManager() {
+		if (this.authenticationManager != null) {
+			return this.authenticationManager;
+		}
+		if (this.reactiveUserDetailsService != null) {
+			UserDetailsRepositoryReactiveAuthenticationManager manager =
+					new UserDetailsRepositoryReactiveAuthenticationManager(this.reactiveUserDetailsService);
+			if (this.passwordEncoder != null) {
+				manager.setPasswordEncoder(this.passwordEncoder);
+			}
+			return manager;
+		}
+		return null;
+	}
+}

config/src/test/java/org/springframework/security/config/annotation/rsocket/HelloHandler.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.security.config.annotation.rsocket;
+
+import io.rsocket.AbstractRSocket;
+import io.rsocket.ConnectionSetupPayload;
+import io.rsocket.Payload;
+import io.rsocket.RSocket;
+import io.rsocket.SocketAcceptor;
+import io.rsocket.util.ByteBufPayload;
+import reactor.core.publisher.Mono;
+
+public class HelloHandler implements SocketAcceptor {
+
+	@Override
+	public Mono<RSocket> accept(ConnectionSetupPayload setup, RSocket sendingSocket) {
+		return Mono.just(
+				new AbstractRSocket() {
+					@Override
+					public Mono<Payload> requestResponse(Payload payload) {
+						String data = payload.getDataUtf8();
+						payload.release();
+						System.out.println("Got " + data);
+						return Mono.just(ByteBufPayload.create("Hello " + data));
+					}
+				});
+	}
+}

config/src/test/java/org/springframework/security/config/annotation/rsocket/JwtITests.java

@@ -0,0 +1,182 @@
+/*
+ * Copyright 2002-2013 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.security.config.annotation.rsocket;
+
+import io.rsocket.RSocketFactory;
+import io.rsocket.frame.decoder.PayloadDecoder;
+import io.rsocket.transport.netty.server.CloseableChannel;
+import io.rsocket.transport.netty.server.TcpServerTransport;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.messaging.handler.annotation.MessageMapping;
+import org.springframework.messaging.rsocket.RSocketRequester;
+import org.springframework.messaging.rsocket.RSocketStrategies;
+import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
+import org.springframework.security.config.Customizer;
+import org.springframework.security.oauth2.core.oidc.IdTokenClaimNames;
+import org.springframework.security.oauth2.jwt.Jwt;
+import org.springframework.security.oauth2.jwt.ReactiveJwtDecoder;
+import org.springframework.security.rsocket.interceptor.PayloadSocketAcceptorInterceptor;
+import org.springframework.security.rsocket.metadata.BasicAuthenticationEncoder;
+import org.springframework.security.rsocket.metadata.BearerTokenMetadata;
+import org.springframework.stereotype.Controller;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringRunner;
+import reactor.core.publisher.Mono;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * @author Rob Winch
+ */
+@ContextConfiguration
+@RunWith(SpringRunner.class)
+public class JwtITests {
+	@Autowired
+	RSocketMessageHandler handler;
+
+	@Autowired
+	PayloadSocketAcceptorInterceptor interceptor;
+
+	@Autowired
+	ServerController controller;
+
+	@Autowired
+	ReactiveJwtDecoder decoder;
+
+	private CloseableChannel server;
+
+	private RSocketRequester requester;
+
+	@Before
+	public void setup() {
+		this.server = RSocketFactory.receive()
+				.frameDecoder(PayloadDecoder.ZERO_COPY)
+				.addSocketAcceptorPlugin(this.interceptor)
+				.acceptor(this.handler.responder())
+				.transport(TcpServerTransport.create("localhost", 7000))
+				.start()
+				.block();
+	}
+
+	@After
+	public void dispose() {
+		this.requester.rsocket().dispose();
+		this.server.dispose();
+		this.controller.payloads.clear();
+	}
+
+	@Test
+	public void routeWhenAuthorized() {
+		BearerTokenMetadata credentials =
+				new BearerTokenMetadata("token");
+		when(this.decoder.decode(any())).thenReturn(Mono.just(jwt()));
+		this.requester = requester()
+				.setupMetadata(credentials.getToken(), BearerTokenMetadata.BEARER_AUTHENTICATION_MIME_TYPE)
+				.connectTcp(this.server.address().getHostName(), this.server.address().getPort())
+				.block();
+
+		String hiRob = this.requester.route("secure.retrieve-mono")
+				.data("rob")
+				.retrieveMono(String.class)
+				.block();
+
+		assertThat(hiRob).isEqualTo("Hi rob");
+	}
+
+	private Jwt jwt() {
+		Map<String, Object> claims = new HashMap<>();
+		claims.put(IdTokenClaimNames.ISS, "https://issuer.example.com");
+		claims.put(IdTokenClaimNames.SUB, "rob");
+		claims.put(IdTokenClaimNames.AUD, Arrays.asList("client-id"));
+		Instant issuedAt = Instant.now();
+		Instant expiresAt = Instant.from(issuedAt).plusSeconds(3600);
+		return new Jwt("token", issuedAt, expiresAt, claims, claims);
+	}
+
+	private RSocketRequester.Builder requester() {
+		return RSocketRequester.builder()
+				.rsocketStrategies(this.handler.getRSocketStrategies());
+	}
+
+
+	@Configuration
+	@EnableRSocketSecurity
+	static class Config {
+
+		@Bean
+		public ServerController controller() {
+			return new ServerController();
+		}
+
+		@Bean
+		public RSocketMessageHandler messageHandler() {
+			RSocketMessageHandler handler = new RSocketMessageHandler();
+			handler.setRSocketStrategies(rsocketStrategies());
+			return handler;
+		}
+
+		@Bean
+		public RSocketStrategies rsocketStrategies() {
+			return RSocketStrategies.builder()
+					.encoder(new BasicAuthenticationEncoder())
+					.build();
+		}
+
+		@Bean
+		PayloadSocketAcceptorInterceptor rsocketInterceptor(RSocketSecurity rsocket) {
+			rsocket
+				.authorizePayload(authorize ->
+					authorize
+						.route("secure.admin.*").authenticated()
+						.anyRequest().permitAll()
+				)
+				.jwt(Customizer.withDefaults());
+			return rsocket.build();
+		}
+
+		@Bean
+		ReactiveJwtDecoder jwtDecoder() {
+			return mock(ReactiveJwtDecoder.class);
+		}
+	}
+
+	@Controller
+	static class ServerController {
+		private List<String> payloads = new ArrayList<>();
+
+		@MessageMapping("**")
+		String connect(String payload) {
+			return "Hi " + payload;
+		}
+	}
+
+}

config/src/test/java/org/springframework/security/config/annotation/rsocket/RSocketMessageHandlerConnectionITests.java

@@ -0,0 +1,246 @@
+/*
+ * Copyright 2002-2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.security.config.annotation.rsocket;
+
+import io.rsocket.RSocketFactory;
+import io.rsocket.exceptions.ApplicationErrorException;
+import io.rsocket.frame.decoder.PayloadDecoder;
+import io.rsocket.transport.netty.server.CloseableChannel;
+import io.rsocket.transport.netty.server.TcpServerTransport;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.messaging.handler.annotation.MessageMapping;
+import org.springframework.messaging.rsocket.RSocketRequester;
+import org.springframework.messaging.rsocket.RSocketStrategies;
+import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
+import org.springframework.security.config.Customizer;
+import org.springframework.security.config.annotation.rsocket.EnableRSocketSecurity;
+import org.springframework.security.config.annotation.rsocket.RSocketSecurity;
+import org.springframework.security.core.userdetails.MapReactiveUserDetailsService;
+import org.springframework.security.core.userdetails.User;
+import org.springframework.security.core.userdetails.UserDetails;
+import org.springframework.security.rsocket.interceptor.PayloadSocketAcceptorInterceptor;
+import org.springframework.security.rsocket.metadata.BasicAuthenticationEncoder;
+import org.springframework.security.rsocket.metadata.UsernamePasswordMetadata;
+import org.springframework.stereotype.Controller;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+/**
+ * @author Rob Winch
+ */
+@ContextConfiguration
+@RunWith(SpringRunner.class)
+public class RSocketMessageHandlerConnectionITests {
+	@Autowired
+	RSocketMessageHandler handler;
+
+	@Autowired
+	PayloadSocketAcceptorInterceptor interceptor;
+
+	@Autowired
+	ServerController controller;
+
+	private CloseableChannel server;
+
+	private RSocketRequester requester;
+
+	@Before
+	public void setup() {
+		this.server = RSocketFactory.receive()
+				.frameDecoder(PayloadDecoder.ZERO_COPY)
+				.addSocketAcceptorPlugin(this.interceptor)
+				.acceptor(this.handler.responder())
+				.transport(TcpServerTransport.create("localhost", 7000))
+				.start()
+				.block();
+	}
+
+	@After
+	public void dispose() {
+		this.requester.rsocket().dispose();
+		this.server.dispose();
+		this.controller.payloads.clear();
+	}
+
+	@Test
+	public void routeWhenAuthorized() {
+		UsernamePasswordMetadata credentials =
+				new UsernamePasswordMetadata("user", "password");
+		this.requester = requester()
+				.setupMetadata(credentials, UsernamePasswordMetadata.BASIC_AUTHENTICATION_MIME_TYPE)
+				.connectTcp(this.server.address().getHostName(), this.server.address().getPort())
+				.block();
+
+		String hiRob = this.requester.route("secure.retrieve-mono")
+				.data("rob")
+				.retrieveMono(String.class)
+				.block();
+
+		assertThat(hiRob).isEqualTo("Hi rob");
+	}
+
+	@Test
+	public void routeWhenNotAuthorized() {
+		UsernamePasswordMetadata credentials = new UsernamePasswordMetadata("user", "password");
+		this.requester = requester()
+				.setupMetadata(credentials, UsernamePasswordMetadata.BASIC_AUTHENTICATION_MIME_TYPE)
+				.connectTcp(this.server.address().getHostName(), this.server.address().getPort())
+				.block();
+
+		assertThatCode(() -> this.requester.route("secure.admin.retrieve-mono")
+				.data("data")
+				.retrieveMono(String.class)
+				.block())
+			.isInstanceOf(ApplicationErrorException.class);
+	}
+
+	@Test
+	public void routeWhenStreamCredentialsAuthorized() {
+		UsernamePasswordMetadata connectCredentials = new UsernamePasswordMetadata("user", "password");
+		this.requester = requester()
+				.setupMetadata(connectCredentials, UsernamePasswordMetadata.BASIC_AUTHENTICATION_MIME_TYPE)
+				.connectTcp(this.server.address().getHostName(), this.server.address().getPort())
+				.block();
+
+		String hiRob = this.requester.route("secure.admin.retrieve-mono")
+				.metadata(new UsernamePasswordMetadata("admin", "password"), UsernamePasswordMetadata.BASIC_AUTHENTICATION_MIME_TYPE)
+				.data("rob")
+				.retrieveMono(String.class)
+				.block();
+
+		assertThat(hiRob).isEqualTo("Hi rob");
+	}
+
+	@Test
+	public void connectWhenNotAuthenticated() {
+		this.requester = requester()
+				.connectTcp(this.server.address().getHostName(), this.server.address().getPort())
+				.block();
+
+		assertThatCode(() -> this.requester.route("retrieve-mono")
+				.data("data")
+				.retrieveMono(String.class)
+				.block())
+				.isNotNull();
+		// FIXME: https://github.com/rsocket/rsocket-java/issues/686
+		//			.isInstanceOf(RejectedSetupException.class);
+	}
+
+	@Test
+	public void connectWhenNotAuthorized() {
+		UsernamePasswordMetadata credentials = new UsernamePasswordMetadata("evil", "password");
+		this.requester = requester()
+				.setupMetadata(credentials, UsernamePasswordMetadata.BASIC_AUTHENTICATION_MIME_TYPE)
+				.connectTcp(this.server.address().getHostName(), this.server.address().getPort())
+				.block();
+
+		assertThatCode(() -> this.requester.route("retrieve-mono")
+				.data("data")
+				.retrieveMono(String.class)
+				.block())
+			.isNotNull();
+//		 FIXME: https://github.com/rsocket/rsocket-java/issues/686
+//			.isInstanceOf(RejectedSetupException.class);
+	}
+
+	private RSocketRequester.Builder requester() {
+		return RSocketRequester.builder()
+				.rsocketStrategies(this.handler.getRSocketStrategies());
+	}
+
+
+	@Configuration
+	@EnableRSocketSecurity
+	static class Config {
+
+		@Bean
+		public ServerController controller() {
+			return new ServerController();
+		}
+
+		@Bean
+		public RSocketMessageHandler messageHandler() {
+			RSocketMessageHandler handler = new RSocketMessageHandler();
+			handler.setRSocketStrategies(rsocketStrategies());
+			return handler;
+		}
+
+		@Bean
+		public RSocketStrategies rsocketStrategies() {
+			return RSocketStrategies.builder()
+					.encoder(new BasicAuthenticationEncoder())
+					.build();
+		}
+
+		@Bean
+		MapReactiveUserDetailsService uds() {
+			UserDetails admin = User.withDefaultPasswordEncoder()
+					.username("admin")
+					.password("password")
+					.roles("USER", "ADMIN", "SETUP")
+					.build();
+			UserDetails user = User.withDefaultPasswordEncoder()
+					.username("user")
+					.password("password")
+					.roles("USER", "SETUP")
+					.build();
+
+			UserDetails evil = User.withDefaultPasswordEncoder()
+					.username("evil")
+					.password("password")
+					.roles("EVIL")
+					.build();
+			return new MapReactiveUserDetailsService(admin, user, evil);
+		}
+
+		@Bean
+		PayloadSocketAcceptorInterceptor rsocketInterceptor(RSocketSecurity rsocket) {
+			rsocket
+				.authorizePayload(authorize ->
+					authorize
+						.setup().hasRole("SETUP")
+						.route("secure.admin.*").hasRole("ADMIN")
+						.route("secure.**").hasRole("USER")
+						.anyRequest().permitAll()
+				)
+				.basicAuthentication(Customizer.withDefaults());
+			return rsocket.build();
+		}
+	}
+
+	@Controller
+	static class ServerController {
+		private List<String> payloads = new ArrayList<>();
+
+		@MessageMapping("**")
+		String connect(String payload) {
+			return "Hi " + payload;
+		}
+	}
+
+}

config/src/test/java/org/springframework/security/config/annotation/rsocket/RSocketMessageHandlerITests.java

@@ -0,0 +1,312 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.security.config.annotation.rsocket;
+
+import io.rsocket.RSocketFactory;
+import io.rsocket.exceptions.ApplicationErrorException;
+import io.rsocket.frame.decoder.PayloadDecoder;
+import io.rsocket.transport.netty.server.CloseableChannel;
+import io.rsocket.transport.netty.server.TcpServerTransport;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.messaging.handler.annotation.MessageMapping;
+import org.springframework.messaging.rsocket.RSocketRequester;
+import org.springframework.messaging.rsocket.RSocketStrategies;
+import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
+import org.springframework.security.config.Customizer;
+import org.springframework.security.config.annotation.rsocket.EnableRSocketSecurity;
+import org.springframework.security.config.annotation.rsocket.RSocketSecurity;
+import org.springframework.security.core.userdetails.MapReactiveUserDetailsService;
+import org.springframework.security.core.userdetails.User;
+import org.springframework.security.core.userdetails.UserDetails;
+import org.springframework.security.rsocket.interceptor.PayloadSocketAcceptorInterceptor;
+import org.springframework.security.rsocket.metadata.BasicAuthenticationEncoder;
+import org.springframework.security.rsocket.metadata.UsernamePasswordMetadata;
+import org.springframework.stereotype.Controller;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringRunner;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+/**
+ * @author Rob Winch
+ */
+@ContextConfiguration
+@RunWith(SpringRunner.class)
+public class RSocketMessageHandlerITests {
+	@Autowired
+	RSocketMessageHandler handler;
+
+	@Autowired
+	PayloadSocketAcceptorInterceptor interceptor;
+
+	@Autowired
+	ServerController controller;
+
+	private CloseableChannel server;
+
+	private RSocketRequester requester;
+
+	@Before
+	public void setup() {
+		this.server = RSocketFactory.receive()
+				.frameDecoder(PayloadDecoder.ZERO_COPY)
+				.addSocketAcceptorPlugin(this.interceptor)
+				.acceptor(this.handler.responder())
+				.transport(TcpServerTransport.create("localhost", 7000))
+				.start()
+				.block();
+
+		this.requester = RSocketRequester.builder()
+				//				.rsocketFactory(factory -> factory.addRequesterPlugin(payloadInterceptor))
+				.rsocketStrategies(this.handler.getRSocketStrategies())
+				.connectTcp("localhost", 7000)
+				.block();
+	}
+
+	@After
+	public void dispose() {
+		this.requester.rsocket().dispose();
+		this.server.dispose();
+		this.controller.payloads.clear();
+	}
+
+	@Test
+	public void retrieveMonoWhenSecureThenDenied() throws Exception {
+		String data = "rob";
+		assertThatCode(() -> this.requester.route("secure.retrieve-mono")
+				.data(data)
+				.retrieveMono(String.class)
+				.block()
+			).isInstanceOf(ApplicationErrorException.class);
+		assertThat(this.controller.payloads).isEmpty();
+	}
+
+	@Test
+	public void retrieveMonoWhenAuthenticationFailedThenException() throws Exception {
+		String data = "rob";
+		UsernamePasswordMetadata credentials = new UsernamePasswordMetadata("invalid", "password");
+		assertThatCode(() -> this.requester.route("secure.retrieve-mono")
+				.metadata(credentials, UsernamePasswordMetadata.BASIC_AUTHENTICATION_MIME_TYPE)
+				.data(data)
+				.retrieveMono(String.class)
+				.block()
+		).isInstanceOf(ApplicationErrorException.class);
+		assertThat(this.controller.payloads).isEmpty();
+	}
+
+	@Test
+	public void retrieveMonoWhenAuthorizedThenGranted() throws Exception {
+		String data = "rob";
+		UsernamePasswordMetadata credentials = new UsernamePasswordMetadata("rob", "password");
+		String hiRob = this.requester.route("secure.retrieve-mono")
+				.metadata(credentials, UsernamePasswordMetadata.BASIC_AUTHENTICATION_MIME_TYPE)
+				.data(data)
+				.retrieveMono(String.class)
+				.block();
+
+		assertThat(hiRob).isEqualTo("Hi rob");
+		assertThat(this.controller.payloads).containsOnly(data);
+	}
+
+	@Test
+	public void retrieveMonoWhenPublicThenGranted() throws Exception {
+		String data = "rob";
+		String hiRob = this.requester.route("retrieve-mono")
+				.data(data)
+				.retrieveMono(String.class)
+				.block();
+
+		assertThat(hiRob).isEqualTo("Hi rob");
+		assertThat(this.controller.payloads).containsOnly(data);
+	}
+
+	@Test
+	public void retrieveFluxWhenDataFluxAndSecureThenDenied() throws Exception {
+		Flux<String> data = Flux.just("a", "b", "c");
+		assertThatCode(() -> this.requester.route("secure.secure.retrieve-flux")
+				.data(data, String.class)
+				.retrieveFlux(String.class)
+				.collectList()
+				.block()).isInstanceOf(
+				ApplicationErrorException.class);
+
+		assertThat(this.controller.payloads).isEmpty();
+	}
+
+	@Test
+	public void retrieveFluxWhenDataFluxAndPublicThenGranted() throws Exception {
+		Flux<String> data = Flux.just("a", "b", "c");
+		List<String> hi = this.requester.route("retrieve-flux")
+				.data(data, String.class)
+				.retrieveFlux(String.class)
+				.collectList()
+				.block();
+
+		assertThat(hi).containsOnly("hello a", "hello b", "hello c");
+		assertThat(this.controller.payloads).containsOnlyElementsOf(data.collectList().block());
+	}
+
+	@Test
+	public void retrieveFluxWhenDataStringAndSecureThenDenied() throws Exception {
+		String data = "a";
+		assertThatCode(() -> this.requester.route("secure.hello")
+				.data(data)
+				.retrieveFlux(String.class)
+				.collectList()
+				.block()).isInstanceOf(
+				ApplicationErrorException.class);
+
+		assertThat(this.controller.payloads).isEmpty();
+	}
+
+	@Test
+	public void retrieveFluxWhenDataStringAndPublicThenGranted() throws Exception {
+		String data = "a";
+		List<String> hi = this.requester.route("retrieve-flux")
+				.data(data)
+				.retrieveFlux(String.class)
+				.collectList()
+				.block();
+
+		assertThat(hi).contains("hello a");
+		assertThat(this.controller.payloads).containsOnly(data);
+	}
+
+	@Test
+	public void sendWhenSecureThenDenied() throws Exception {
+		String data = "hi";
+		this.requester.route("secure.send")
+				.data(data)
+				.send()
+				.block();
+
+		assertThat(this.controller.payloads).isEmpty();
+	}
+
+	@Test
+	public void sendWhenPublicThenGranted() throws Exception {
+		String data = "hi";
+		this.requester.route("send")
+				.data(data)
+				.send()
+				.block();
+		assertThat(this.controller.awaitPayloads()).containsOnly("hi");
+	}
+
+	@Configuration
+	@EnableRSocketSecurity
+	static class Config {
+
+		@Bean
+		public ServerController controller() {
+			return new ServerController();
+		}
+
+		@Bean
+		public RSocketMessageHandler messageHandler() {
+			RSocketMessageHandler handler = new RSocketMessageHandler();
+			handler.setRSocketStrategies(rsocketStrategies());
+			return handler;
+		}
+
+		@Bean
+		public RSocketStrategies rsocketStrategies() {
+			return RSocketStrategies.builder()
+					.encoder(new BasicAuthenticationEncoder())
+					.build();
+		}
+
+		@Bean
+		MapReactiveUserDetailsService uds() {
+			UserDetails rob = User.withDefaultPasswordEncoder()
+					.username("rob")
+					.password("password")
+					.roles("USER", "ADMIN")
+					.build();
+			UserDetails rossen = User.withDefaultPasswordEncoder()
+					.username("rossen")
+					.password("password")
+					.roles("USER")
+					.build();
+			return new MapReactiveUserDetailsService(rob, rossen);
+		}
+
+		@Bean
+		PayloadSocketAcceptorInterceptor rsocketInterceptor(RSocketSecurity rsocket) {
+			rsocket
+					.authorizePayload(authorize -> {
+						authorize
+								.route("secure.*").authenticated()
+								.anyRequest().permitAll();
+					})
+					.basicAuthentication(Customizer.withDefaults());
+			return rsocket.build();
+		}
+	}
+
+	@Controller
+	static class ServerController {
+		private List<String> payloads = new ArrayList<>();
+
+		@MessageMapping({"secure.retrieve-mono", "retrieve-mono"})
+		String retrieveMono(String payload) {
+			add(payload);
+			return "Hi " + payload;
+		}
+
+		@MessageMapping({"secure.retrieve-flux", "retrieve-flux"})
+		Flux<String> retrieveFlux(Flux<String> payload) {
+			return payload.doOnNext(this::add)
+					.map(p -> "hello " + p);
+		}
+
+		@MessageMapping({"secure.send", "send"})
+		Mono<Void> send(Flux<String> payload) {
+			return payload
+					.doOnNext(this::add)
+					.then(Mono.fromRunnable(() -> {
+						doNotifyAll();
+					}));
+		}
+
+		private synchronized void doNotifyAll() {
+			this.notifyAll();
+		}
+
+		private synchronized List<String> awaitPayloads() throws InterruptedException {
+			this.wait();
+			return this.payloads;
+		}
+
+		private void add(String p) {
+			this.payloads.add(p);
+		}
+	}
+
+}