Spring Integration中的安全性

1. 简介

本文将重点介绍如何在Spring Integration的集成流程中结合使用Spring Security。我们将搭建一个简单的安全消息流来演示Spring Security在Spring Integration中的应用，并提供多线程消息通道中SecurityContext传播的示例。

关于框架的更多使用细节，可参考我们的Spring Integration入门指南。

2. Spring Integration配置

2.1 依赖项

首先，我们需要在项目中添加Spring Integration依赖。由于我们将使用DirectChannel、PublishSubscribeChannel和ServiceActivator搭建简单消息流，需要spring-integration-core依赖：

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-core</artifactId>
    <version>6.0.0</version>
</dependency>

同时，为了在Spring Integration中使用Spring Security，需要添加spring-integration-security依赖：

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-security</artifactId>
    <version>6.0.0</version>
</dependency>

此外，我们使用Spring Security，因此添加spring-security-config：

<dependency>
    <groupId>org.springframework.security</groupId>
    <artifactId>spring-security-config</artifactId>
    <version>6.0.0</version>
</dependency>

所有依赖的最新版本可在Maven Central查看：spring-integration-security，spring-security-config。

2.2 基于Java的配置

我们的示例将使用基本的Spring Integration组件。只需通过@EnableIntegration注解启用Spring Integration：

@Configuration
@EnableIntegration
public class SecuredDirectChannel {
    //...
}

3. 安全消息通道

关键点：需要ChannelSecurityInterceptor实例拦截通道上的所有send和receive调用，并决定是否允许执行：

@Autowired
@Bean
public ChannelSecurityInterceptor channelSecurityInterceptor(
  AuthenticationManager authenticationManager, 
  AccessDecisionManager customAccessDecisionManager) {

    ChannelSecurityInterceptor 
      channelSecurityInterceptor = new ChannelSecurityInterceptor();

    channelSecurityInterceptor
      .setAuthenticationManager(authenticationManager);

    channelSecurityInterceptor
      .setAccessDecisionManager(customAccessDecisionManager);

    return channelSecurityInterceptor;
}

AuthenticationManager和AccessDecisionManagerBean定义如下：

@Configuration
@EnableGlobalMethodSecurity(prePostEnabled = true)
public class SecurityConfig extends GlobalMethodSecurityConfiguration {

    @Override
    @Bean
    public AuthenticationManager 
      authenticationManager() throws Exception {
        return super.authenticationManager();
    }

    @Bean
    public AccessDecisionManager customAccessDecisionManager() {
        List<AccessDecisionVoter<? extends Object>> 
          decisionVoters = new ArrayList<>();
        decisionVoters.add(new RoleVoter());
        decisionVoters.add(new UsernameAccessDecisionVoter());
        AccessDecisionManager accessDecisionManager
          = new AffirmativeBased(decisionVoters);
        return accessDecisionManager;
    }
}

这里使用了两个AccessDecisionVoter：RoleVoter和自定义的UsernameAccessDecisionVoter。

现在使用ChannelSecurityInterceptor保护通道，通过@SecuredChannel注解装饰通道：

@Bean(name = "startDirectChannel")
@SecuredChannel(
  interceptor = "channelSecurityInterceptor", 
  sendAccess = { "ROLE_VIEWER","jane" })
public DirectChannel startDirectChannel() {
    return new DirectChannel();
}

@Bean(name = "endDirectChannel")
@SecuredChannel(
  interceptor = "channelSecurityInterceptor", 
  sendAccess = {"ROLE_EDITOR"})
public DirectChannel endDirectChannel() {
    return new DirectChannel();
}

@SecuredChannel接受三个属性：

interceptor：引用ChannelSecurityInterceptor Bean
sendAccess和receiveAccess：定义调用通道send或receive操作的权限策略

示例中，只有拥有ROLE_VIEWER或用户名为jane的用户才能向startDirectChannel发送消息，只有拥有ROLE_EDITOR的用户才能向endDirectChannel发送消息。这通过自定义AccessDecisionManager实现：只要RoleVoter或UsernameAccessDecisionVoter任一返回肯定响应即授权通过。

4. 安全的ServiceActivator

注意：我们也可以使用Spring方法安全保护ServiceActivator。需要启用方法安全注解：

@Configuration
@EnableGlobalMethodSecurity(prePostEnabled = true)
public class SecurityConfig extends GlobalMethodSecurityConfiguration {
    //....
}

为简化起见，本文仅使用Spring的pre和post注解，因此在配置类添加@EnableGlobalMethodSecurity并设置prePostEnabled=true。

现在使用@PreAuthorization注解保护ServiceActivator：

@ServiceActivator(
  inputChannel = "startDirectChannel", 
  outputChannel = "endDirectChannel")
@PreAuthorize("hasRole('ROLE_LOGGER')")
public Message<?> logMessage(Message<?> message) {
    Logger.getAnonymousLogger().info(message.toString());
    return message;
}

此ServiceActivator从startDirectChannel接收消息并输出到endDirectChannel。仅当当前Authentication主体拥有ROLE_LOGGER角色时才能访问该方法。

5. 安全上下文传播

Spring的SecurityContext默认绑定到线程。这意味着SecurityContext不会传播到子线程。

上述示例使用DirectChannel和ServiceActivator——它们都在单线程中运行，因此SecurityContext在整个流程中可用。

但当使用QueueChannel、ExecutorChannel或带Executor的PublishSubscribeChannel时，消息会从一个线程传递到其他线程。此时需要将SecurityContext传播到所有接收消息的线程。

创建另一个消息流，以PublishSubscribeChannel开始，两个ServiceActivator订阅该通道：

@Bean(name = "startPSChannel")
@SecuredChannel(
  interceptor = "channelSecurityInterceptor", 
  sendAccess = "ROLE_VIEWER")
public PublishSubscribeChannel startChannel() {
    return new PublishSubscribeChannel(executor());
}

@ServiceActivator(
  inputChannel = "startPSChannel", 
  outputChannel = "finalPSResult")
@PreAuthorize("hasRole('ROLE_LOGGER')")
public Message<?> changeMessageToRole(Message<?> message) {
    return buildNewMessage(getRoles(), message);
}

@ServiceActivator(
  inputChannel = "startPSChannel", 
  outputChannel = "finalPSResult")
@PreAuthorize("hasRole('ROLE_VIEWER')")
public Message<?> changeMessageToUserName(Message<?> message) {
    return buildNewMessage(getUsername(), message);
}

示例中有两个ServiceActivator订阅startPSChannel。该通道要求Authentication主体拥有ROLE_VIEWER角色才能发送消息。同样，仅当Authentication主体拥有ROLE_LOGGER角色时才能调用changeMessageToRole服务，而changeMessageToUserName服务仅当主体拥有ROLE_VIEWER角色时才能调用。

startPSChannel将运行在ThreadPoolTaskExecutor支持下：

@Bean
public ThreadPoolTaskExecutor executor() {
    ThreadPoolTaskExecutor pool = new ThreadPoolTaskExecutor();
    pool.setCorePoolSize(10);
    pool.setMaxPoolSize(10);
    pool.setWaitForTasksToCompleteOnShutdown(true);
    return pool;
}

因此，两个ServiceActivator将在不同线程中运行。**要将SecurityContext传播到这些线程，需要在消息通道添加SecurityContextPropagationChannelInterceptor**：

@Bean
@GlobalChannelInterceptor(patterns = { "startPSChannel" })
public ChannelInterceptor securityContextPropagationInterceptor() {
    return new SecurityContextPropagationChannelInterceptor();
}

注意如何用@GlobalChannelInterceptor注解装饰SecurityContextPropagationChannelInterceptor，并在其patterns属性中添加startPSChannel。此配置表示当前线程的SecurityContext将传播到从startPSChannel派生的所有线程。

6. 测试

使用JUnit测试验证消息流。

6.1 依赖

此时需要spring-security-test依赖：

<dependency>
    <groupId>org.springframework.security</groupId>
    <artifactId>spring-security-test</artifactId>
    <version>6.0.0</version>
    <scope>test</scope>
</dependency>

最新版本可在Maven Central查看：spring-security-test。

6.2 测试安全通道

首先尝试向startDirectChannel发送消息：

@Test(expected = AuthenticationCredentialsNotFoundException.class)
public void 
  givenNoUser_whenSendToDirectChannel_thenCredentialNotFound() {

    startDirectChannel
      .send(new GenericMessage<String>(DIRECT_CHANNEL_MESSAGE));
}

由于通道受保护，未提供认证对象时发送消息会抛出AuthenticationCredentialsNotFoundException异常。

接下来提供拥有ROLE_VIEWER角色的用户，向startDirectChannel发送消息：

@Test
@WithMockUser(roles = { "VIEWER" })
public void 
  givenRoleViewer_whenSendToDirectChannel_thenAccessDenied() {
    expectedException.expectCause
      (IsInstanceOf.<Throwable> instanceOf(AccessDeniedException.class));

    startDirectChannel
      .send(new GenericMessage<String>(DIRECT_CHANNEL_MESSAGE));
 }

虽然用户因拥有ROLE_VIEWER角色可向startDirectChannel发送消息，但无法调用需要ROLE_LOGGER角色的logMessage服务。此时将抛出包含AccessDeniedException原因的MessageHandlingException。

现在提供用户名为jane且拥有ROLE_LOGGER和ROLE_EDITOR角色的用户：

@Test
@WithMockUser(username = "jane", roles = { "LOGGER", "EDITOR" })
public void 
  givenJaneLoggerEditor_whenSendToDirectChannel_thenFlowCompleted() {
    startDirectChannel
      .send(new GenericMessage<String>(DIRECT_CHANNEL_MESSAGE));
    assertEquals
      (DIRECT_CHANNEL_MESSAGE, messageConsumer.getMessageContent());
}

消息将成功流经整个流程：从startDirectChannel到logMessage激活器，再到endDirectChannel。因为提供的认证对象拥有访问这些组件所需的所有权限。

6.3 测试安全上下文传播

回顾使用PublishSubscribeChannel的完整流程：

流程从startPSChannel开始，策略为sendAccess = "ROLE_VIEWER"
两个ServiceActivator订阅该通道：一个有@PreAuthorize("hasRole('ROLE_LOGGER')")注解，另一个有@PreAuthorize("hasRole('ROLE_VIEWER')")注解

首先提供拥有ROLE_VIEWER角色的用户，尝试向通道发送消息：

@Test
@WithMockUser(username = "user", roles = { "VIEWER" })
public void 
  givenRoleUser_whenSendMessageToPSChannel_thenNoMessageArrived() 
  throws IllegalStateException, InterruptedException {
 
    startPSChannel
      .send(new GenericMessage<String>(DIRECT_CHANNEL_MESSAGE));

    executor
      .getThreadPoolExecutor()
      .awaitTermination(2, TimeUnit.SECONDS);

    assertEquals(1, messageConsumer.getMessagePSContent().size());
    assertTrue(
      messageConsumer
      .getMessagePSContent().values().contains("user"));
}

**由于用户仅拥有ROLE_VIEWER角色，消息只能通过startPSChannel和一个ServiceActivator**。流程结束时只收到一条消息。

现在提供同时拥有ROLE_VIEWER和ROLE_LOGGER角色的用户：

@Test
@WithMockUser(username = "user", roles = { "LOGGER", "VIEWER" })
public void 
  givenRoleUserAndLogger_whenSendMessageToPSChannel_then2GetMessages() 
  throws IllegalStateException, InterruptedException {
    startPSChannel
      .send(new GenericMessage<String>(DIRECT_CHANNEL_MESSAGE));

    executor
      .getThreadPoolExecutor()
      .awaitTermination(2, TimeUnit.SECONDS);

    assertEquals(2, messageConsumer.getMessagePSContent().size());
    assertTrue
      (messageConsumer
      .getMessagePSContent()
      .values().contains("user"));
    assertTrue
      (messageConsumer
      .getMessagePSContent()
      .values().contains("ROLE_LOGGER,ROLE_VIEWER"));
}

由于用户拥有所有必需权限，流程结束时收到两条消息。

7. 结论

本文探讨了在Spring Integration中使用Spring Security保护消息通道和ServiceActivator的可能性。

所有示例代码可在GitHub获取。

Persistence

REST

Security