Fix PreAuthorize when returning Kotlin Flow

Closes gh-9676

config/spring-security-config.gradle

@@ -87,6 +87,7 @@ dependencies {
 	}
 	testImplementation 'org.jetbrains.kotlinx:kotlinx-coroutines-core'
 	testImplementation 'org.jetbrains.kotlinx:kotlinx-coroutines-reactor'
+	testImplementation 'io.mockk:mockk'
 
 	testRuntimeOnly 'org.hsqldb:hsqldb'
 }

config/src/test/kotlin/org/springframework/security/config/annotation/method/configuration/KotlinEnableReactiveMethodSecurityTests.kt

@@ -16,11 +16,16 @@
 
 package org.springframework.security.config.annotation.method.configuration
 
+import io.mockk.Called
+import io.mockk.clearAllMocks
+import io.mockk.mockk
+import io.mockk.verify
 import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.runBlocking
 import org.assertj.core.api.Assertions.assertThat
 import org.assertj.core.api.Assertions.assertThatExceptionOfType
+import org.junit.After
 import org.junit.Test
 import org.junit.runner.RunWith
 import org.springframework.beans.factory.annotation.Autowired
@@ -35,11 +40,23 @@ import org.springframework.test.context.junit4.SpringRunner
 @ContextConfiguration
 class KotlinEnableReactiveMethodSecurityTests {
 
+    private lateinit var delegate: KotlinReactiveMessageService
+
     @Autowired
     var messageService: KotlinReactiveMessageService? = null
 
+    @After
+    fun cleanup() {
+        clearAllMocks()
+    }
+
+    @Autowired
+    fun setConfig(config: Config) {
+        this.delegate = config.delegate
+    }
+
     @Test
-    fun suspendingGetResultWhenPermitAllThenSuccess() {
+    fun `suspendingNoAuth always success`() {
         runBlocking {
             assertThat(messageService!!.suspendingNoAuth()).isEqualTo("success")
         }
@@ -47,14 +64,14 @@ class KotlinEnableReactiveMethodSecurityTests {
 
     @Test
     @WithMockUser(authorities = ["ROLE_ADMIN"])
-    fun suspendingPreAuthorizeHasRoleWhenGrantedThenSuccess() {
+    fun `suspendingPreAuthorizeHasRole when user has role then success`() {
         runBlocking {
             assertThat(messageService!!.suspendingPreAuthorizeHasRole()).isEqualTo("admin")
         }
     }
 
     @Test
-    fun suspendingPreAuthorizeHasRoleWhenNoAuthenticationThenDenied() {
+    fun `suspendingPreAuthorizeHasRole when user does not have role then denied`() {
         assertThatExceptionOfType(AccessDeniedException::class.java).isThrownBy {
             runBlocking {
                 messageService!!.suspendingPreAuthorizeHasRole()
@@ -64,14 +81,14 @@ class KotlinEnableReactiveMethodSecurityTests {
 
     @Test
     @WithMockUser
-    fun suspendingPreAuthorizeBeanWhenGrantedThenSuccess() {
+    fun `suspendingPreAuthorizeBean when authorized then success`() {
         runBlocking {
             assertThat(messageService!!.suspendingPreAuthorizeBean(true)).isEqualTo("check")
         }
     }
 
     @Test
-    fun suspendingPreAuthorizeBeanWhenNotAuthorizedThenDenied() {
+    fun `suspendingPreAuthorizeBean when not authorized then denied`() {
         assertThatExceptionOfType(AccessDeniedException::class.java).isThrownBy {
             runBlocking {
                 messageService!!.suspendingPreAuthorizeBean(false)
@@ -81,7 +98,7 @@ class KotlinEnableReactiveMethodSecurityTests {
 
     @Test
     @WithMockUser("user")
-    fun suspendingPostAuthorizeWhenAuthorizedThenSuccess() {
+    fun `suspendingPostAuthorize when authorized then success`() {
         runBlocking {
             assertThat(messageService!!.suspendingPostAuthorizeContainsName()).isEqualTo("user")
         }
@@ -89,7 +106,7 @@ class KotlinEnableReactiveMethodSecurityTests {
 
     @Test
     @WithMockUser("other-user")
-    fun suspendingPostAuthorizeWhenNotAuthorizedThenDenied() {
+    fun `suspendingPostAuthorize when not authorized then denied`() {
         assertThatExceptionOfType(AccessDeniedException::class.java).isThrownBy {
             runBlocking {
                 messageService!!.suspendingPostAuthorizeContainsName()
@@ -97,16 +114,26 @@ class KotlinEnableReactiveMethodSecurityTests {
         }
     }
 
+    @Test
+    fun `suspendingPreAuthorizeDelegate when user does not have role then delegate not called`() {
+        assertThatExceptionOfType(AccessDeniedException::class.java).isThrownBy {
+            runBlocking {
+                messageService!!.suspendingPreAuthorizeDelegate()
+            }
+        }
+        verify { delegate wasNot Called }
+    }
+
     @Test
     @WithMockUser(authorities = ["ROLE_ADMIN"])
-    fun suspendingFlowPreAuthorizeHasRoleWhenGrantedThenSuccess() {
+    fun `suspendingFlowPreAuthorize when user has role then success`() {
         runBlocking {
             assertThat(messageService!!.suspendingFlowPreAuthorize().toList()).containsExactly(1, 2, 3)
         }
     }
 
     @Test
-    fun suspendingFlowPreAuthorizeHasRoleWhenNoAuthenticationThenDenied() {
+    fun `suspendingFlowPreAuthorize when user does not have role then denied`() {
         assertThatExceptionOfType(AccessDeniedException::class.java).isThrownBy {
             runBlocking {
                 messageService!!.suspendingFlowPreAuthorize().collect()
@@ -115,14 +142,14 @@ class KotlinEnableReactiveMethodSecurityTests {
     }
 
     @Test
-    fun suspendingFlowPostAuthorizeWhenAuthorizedThenSuccess() {
+    fun `suspendingFlowPostAuthorize when authorized then success`() {
         runBlocking {
             assertThat(messageService!!.suspendingFlowPostAuthorize(true).toList()).containsExactly(1, 2, 3)
         }
     }
 
     @Test
-    fun suspendingFlowPostAuthorizeWhenNotAuthorizedThenDenied() {
+    fun `suspendingFlowPostAuthorize when not authorized then denied`() {
         assertThatExceptionOfType(AccessDeniedException::class.java).isThrownBy {
             runBlocking {
                 messageService!!.suspendingFlowPostAuthorize(false).collect()
@@ -130,16 +157,26 @@ class KotlinEnableReactiveMethodSecurityTests {
         }
     }
 
+    @Test
+    fun `suspendingFlowPreAuthorizeDelegate when not authorized then delegate not called`() {
+        assertThatExceptionOfType(AccessDeniedException::class.java).isThrownBy {
+            runBlocking {
+                messageService!!.suspendingFlowPreAuthorizeDelegate().collect()
+            }
+        }
+        verify { delegate wasNot Called }
+    }
+
     @Test
     @WithMockUser(authorities = ["ROLE_ADMIN"])
-    fun flowPreAuthorizeHasRoleWhenGrantedThenSuccess() {
+    fun `flowPreAuthorize when user has role then success`() {
         runBlocking {
             assertThat(messageService!!.flowPreAuthorize().toList()).containsExactly(1, 2, 3)
         }
     }
 
     @Test
-    fun flowPreAuthorizeHasRoleWhenNoAuthenticationThenDenied() {
+    fun `flowPreAuthorize when user does not have role then denied`() {
         assertThatExceptionOfType(AccessDeniedException::class.java).isThrownBy {
             runBlocking {
                 messageService!!.flowPreAuthorize().collect()
@@ -148,14 +185,14 @@ class KotlinEnableReactiveMethodSecurityTests {
     }
 
     @Test
-    fun flowPostAuthorizeWhenAuthorizedThenSuccess() {
+    fun `flowPostAuthorize when authorized then success`() {
         runBlocking {
             assertThat(messageService!!.flowPostAuthorize(true).toList()).containsExactly(1, 2, 3)
         }
     }
 
     @Test
-    fun flowPostAuthorizeWhenNotAuthorizedThenDenied() {
+    fun `flowPostAuthorize when not authorized then denied`() {
         assertThatExceptionOfType(AccessDeniedException::class.java).isThrownBy {
             runBlocking {
                 messageService!!.flowPostAuthorize(false).collect()
@@ -163,13 +200,24 @@ class KotlinEnableReactiveMethodSecurityTests {
         }
     }
 
+    @Test
+    fun `flowPreAuthorizeDelegate when user does not have role then delegate not called`() {
+        assertThatExceptionOfType(AccessDeniedException::class.java).isThrownBy {
+            runBlocking {
+                messageService!!.flowPreAuthorizeDelegate().collect()
+            }
+        }
+        verify { delegate wasNot Called }
+    }
+
     @EnableReactiveMethodSecurity
     @Configuration
     open class Config {
+        var delegate = mockk<KotlinReactiveMessageService>()
 
         @Bean
         open fun messageService(): KotlinReactiveMessageServiceImpl {
-            return KotlinReactiveMessageServiceImpl()
+            return KotlinReactiveMessageServiceImpl(this.delegate)
         }
 
         @Bean

config/src/test/kotlin/org/springframework/security/config/annotation/method/configuration/KotlinReactiveMessageService.kt

@@ -28,11 +28,17 @@ interface KotlinReactiveMessageService {
 
     suspend fun suspendingPostAuthorizeContainsName(): String
 
+    suspend fun suspendingPreAuthorizeDelegate(): String
+
     suspend fun suspendingFlowPreAuthorize(): Flow<Int>
 
     suspend fun suspendingFlowPostAuthorize(id: Boolean): Flow<Int>
 
+    suspend fun suspendingFlowPreAuthorizeDelegate(): Flow<Int>
+
     fun flowPreAuthorize(): Flow<Int>
 
     fun flowPostAuthorize(id: Boolean): Flow<Int>
+
+    fun flowPreAuthorizeDelegate(): Flow<Int>
 }

config/src/test/kotlin/org/springframework/security/config/annotation/method/configuration/KotlinReactiveMessageServiceImpl.kt

@@ -22,7 +22,7 @@ import kotlinx.coroutines.flow.flow
 import org.springframework.security.access.prepost.PostAuthorize
 import org.springframework.security.access.prepost.PreAuthorize
 
-class KotlinReactiveMessageServiceImpl : KotlinReactiveMessageService {
+class KotlinReactiveMessageServiceImpl(val delegate: KotlinReactiveMessageService) : KotlinReactiveMessageService {
 
     override suspend fun suspendingNoAuth(): String {
         delay(1)
@@ -47,6 +47,11 @@ class KotlinReactiveMessageServiceImpl : KotlinReactiveMessageService {
         return "user"
     }
 
+    @PreAuthorize("hasRole('ADMIN')")
+    override suspend fun suspendingPreAuthorizeDelegate(): String {
+        return delegate.suspendingPreAuthorizeHasRole()
+    }
+
     @PreAuthorize("hasRole('ADMIN')")
     override suspend fun suspendingFlowPreAuthorize(): Flow<Int> {
         delay(1)
@@ -69,6 +74,12 @@ class KotlinReactiveMessageServiceImpl : KotlinReactiveMessageService {
         }
     }
 
+    @PreAuthorize("hasRole('ADMIN')")
+    override suspend fun suspendingFlowPreAuthorizeDelegate(): Flow<Int> {
+        delay(1)
+        return delegate.flowPreAuthorize()
+    }
+
     @PreAuthorize("hasRole('ADMIN')")
     override fun flowPreAuthorize(): Flow<Int> {
         return flow {
@@ -88,4 +99,9 @@ class KotlinReactiveMessageServiceImpl : KotlinReactiveMessageService {
             }
         }
     }
+
+    @PreAuthorize("hasRole('ADMIN')")
+    override fun flowPreAuthorizeDelegate(): Flow<Int> {
+        return delegate.flowPreAuthorize()
+    }
 }

core/src/main/java/org/springframework/security/access/prepost/PrePostAdviceReactiveMethodInterceptor.java

@@ -121,19 +121,21 @@ public class PrePostAdviceReactiveMethodInterceptor implements MethodInterceptor
 					.map((r) -> (attr != null) ? this.postAdvice.after(auth, invocation, attr, r) : r));
 		}
 		if (hasFlowReturnType) {
-			Publisher<?> publisher;
+			Flux<?> response;
 			if (isSuspendingFunction) {
-				publisher = CoroutinesUtils.invokeSuspendingFunction(invocation.getMethod(), invocation.getThis(),
-						invocation.getArguments());
+				response = toInvoke.flatMapMany((auth) -> Flux
+						.from(CoroutinesUtils.invokeSuspendingFunction(invocation.getMethod(), invocation.getThis(),
+								invocation.getArguments()))
+						.map((r) -> (attr != null) ? this.postAdvice.after(auth, invocation, attr, r) : r));
 			}
 			else {
 				ReactiveAdapter adapter = ReactiveAdapterRegistry.getSharedInstance().getAdapter(returnType);
 				Assert.state(adapter != null, () -> "The returnType " + returnType + " on " + method
 						+ " must have a org.springframework.core.ReactiveAdapter registered");
-				publisher = adapter.toPublisher(PrePostAdviceReactiveMethodInterceptor.flowProceed(invocation));
-			}
-			Flux<?> response = toInvoke.flatMapMany((auth) -> Flux.from(publisher)
+				response = toInvoke.flatMapMany((auth) -> Flux
+						.from(adapter.toPublisher(PrePostAdviceReactiveMethodInterceptor.flowProceed(invocation)))
 						.map((r) -> (attr != null) ? this.postAdvice.after(auth, invocation, attr, r) : r));
+			}
 			return KotlinDelegate.asFlow(response);
 		}
 		if (isSuspendingFunction) {

dependencies/spring-security-dependencies.gradle

@@ -25,6 +25,7 @@ dependencies {
 		api "commons-codec:commons-codec:1.15"
 		api "commons-collections:commons-collections:3.2.2"
 		api "commons-logging:commons-logging:1.2"
+		api "io.mockk:mockk:1.10.2"
 		api "io.projectreactor.tools:blockhound:1.0.6.RELEASE"
 		api "javax.annotation:jsr250-api:1.0"
 		api "javax.servlet.jsp.jstl:javax.servlet.jsp.jstl-api:1.2.2"