如何为SCDF的http-source-kafka连接器配置OAuth2安全？

http-source-kafka官方连接器没有内置OAuth2相关配置属性，要实现端点访问权限控制，需要通过自定义扩展连接器的方式集成Spring Security OAuth2资源服务器能力，以下是具体实现步骤：

1. 扩展连接器项目 #

基于官方http-source-kafka的代码结构，添加OAuth2资源服务器依赖和自定义配置：

添加依赖 #

在连接器的pom.xml（或build.gradle）中引入Spring Security OAuth2资源服务器依赖：

<!-- Maven -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-oauth2-resource-server</artifactId>
</dependency>

绑定自定义配置属性 #

创建配置类，将你提到的参数绑定为可配置的属性：

import org.springframework.boot.context.properties.ConfigurationProperties;

@ConfigurationProperties(prefix = "http.source.security")
public class HttpSourceSecurityProperties {
    private String[] permittedEndpoints;
    private String[] restrictedEndpoints;
    private String[] scopes;
    private String resourceId;

    // Getters and Setters
}

2. 配置Spring Security OAuth2规则 #

编写Security配置类，利用上述属性实现端点权限控制：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.security.config.annotation.web.builders.HttpSecurity;
import org.springframework.security.config.annotation.web.configurers.oauth2.server.resource.OAuth2ResourceServerConfigurer;
import org.springframework.security.oauth2.core.OAuth2TokenValidator;
import org.springframework.security.oauth2.jwt.Jwt;
import org.springframework.security.oauth2.jwt.JwtAuthenticationConverter;
import org.springframework.security.oauth2.jwt.JwtValidators;
import org.springframework.security.oauth2.server.resource.authentication.JwtAuthenticationConverter;
import org.springframework.security.web.SecurityFilterChain;

@Configuration
public class HttpSourceSecurityConfig {

    private final HttpSourceSecurityProperties securityProps;

    public HttpSourceSecurityConfig(HttpSourceSecurityProperties securityProps) {
        this.securityProps = securityProps;
    }

    @Bean
    public SecurityFilterChain securityFilterChain(HttpSecurity http) throws Exception {
        http.authorizeHttpRequests(auth -> auth
                // 允许匿名访问的端点
                .requestMatchers(securityProps.getPermittedEndpoints()).permitAll()
                // 需要OAuth2令牌且具备指定scope的端点
                .requestMatchers(securityProps.getRestrictedEndpoints()).hasAnyAuthority("SCOPE_" + String.join(",SCOPE_", securityProps.getScopes()))
                // 其他端点默认拒绝
                .anyRequest().denyAll()
        )
        .oauth2ResourceServer(OAuth2ResourceServerConfigurer::jwt); // 若使用JWT令牌，否则用.introspection()

        // 验证resourceId
        if (securityProps.getResourceId() != null) {
            http.oauth2ResourceServer(oauth2 -> oauth2
                    .jwt(jwt -> jwt.jwtAuthenticationConverter(jwtAuthenticationConverter()))
            );
        }

        return http.build();
    }

    // 自定义JWT转换器，验证resourceId并提取权限
    private JwtAuthenticationConverter jwtAuthenticationConverter() {
        JwtAuthenticationConverter converter = new JwtAuthenticationConverter();
        converter.setJwtGrantedAuthoritiesConverter(jwt -> {
            // 验证resourceId是否匹配
            String resourceId = jwt.getClaimAsString("aud");
            if (!securityProps.getResourceId().equals(resourceId)) {
                throw new RuntimeException("Invalid resource ID in token");
            }
            // 提取scope作为权限
            return jwt.getClaimAsStringList("scope")
                    .stream()
                    .map(scope -> "SCOPE_" + scope)
                    .map(org.springframework.security.core.authority.SimpleGrantedAuthority::new)
                    .collect(java.util.stream.Collectors.toList());
        });
        return converter;
    }
}

3. 部署自定义连接器 #

将扩展后的连接器打包为Jar，注册到SCDF服务器中，部署时即可通过以下参数配置安全规则：

http.source.security.permitted-endpoints=/api/public/**
http.source.security.restricted-endpoints=/api/private/**,/api/secure/**
http.source.security.scopes=read,write
http.source.security.resourceId=my-http-source
# OAuth2资源服务器基础配置（根据实际服务器调整）
spring.security.oauth2.resourceserver.jwt.issuer-uri=https://your-oauth2-server.com/auth/realms/your-realm

关键说明 #

• 若你的OAuth2服务器使用令牌 introspection 而非JWT，将配置中的.jwt()替换为.introspection()，并添加spring.security.oauth2.resourceserver.opaquetoken.introspection-uri等相关配置。
• permitted.endpoints和restricted.endpoints支持Ant风格路径匹配，比如/api/**。

内容的提问来源于stack exchange，提问作者Zishnu