我正在使用 Spring Boot 和 Spring Security 开发反应式应用程序，特别是处理 Kafka 消费者中的身份验证。尽管使用 ReactiveSecurityContextHolder.withSecurityContext(Mono.just(securityContext)) 设置安全上下文，但立即访问同一反应链中的上下文会导致空上下文。我正在遵循反应式应用程序的推荐实践。下面是设置和访问上下文的代码片段：
public <T> Mono<Void> authenticateAndVerifyContext(ReceiverRecord<String, T> record) {
return extractCredentialsFromBasicAuth(record)
.flatMap(this::authenticate)
.flatMap(authentication -> {
SecurityContext securityContext = new SecurityContextImpl(authentication);
log.info("securityContext: {}", securityContext);
return Mono.deferContextual(view -> Mono.empty())
.contextWrite(ReactiveSecurityContextHolder.withSecurityContext(Mono.just(securityContext)))
.then(ReactiveSecurityContextHolder.getContext()
.flatMap(context -> {
if (context.getAuthentication() != null && context.getAuthentication().isAuthenticated()) {
log.info("Security context successfully updated. User: {}", context.getAuthentication().getName());
} else {
log.error("Authentication failed");
}
return Mono.empty();
})
.switchIfEmpty(Mono.defer(() -> {
log.error("Security context is empty");
return Mono.empty();
})));
}).then();
}
private <T> Mono<UsernamePasswordAuthenticationToken> extractCredentialsFromBasicAuth(
final ReceiverRecord<String, T> record) {
log.info("kafka authorization: {}",
new String(record.headers().lastHeader(AUTHORIZATION_KEY).value(), StandardCharsets.UTF_8));
return Mono.justOrEmpty(record.headers().lastHeader(AUTHORIZATION_KEY))
.map(header -> new String(header.value(), StandardCharsets.UTF_8))
.map(this::decodeBasicAuthHeader)
.flatMap(this::createAuthenticationToken);
}
private String decodeBasicAuthHeader(final String headerValue) {
return new String(Base64.getDecoder().decode(headerValue.substring(6)), StandardCharsets.UTF_8);
}
private Mono<UsernamePasswordAuthenticationToken> createAuthenticationToken(final String credentials) {
final String[] parts = credentials.split(":", 2);
if (parts.length == 2) {
return Mono.just(new UsernamePasswordAuthenticationToken(parts[0], parts[1]));
} else {
return Mono.error(new IllegalArgumentException("Invalid basic authentication token"));
}
}
尽管 .contextWrite 和 ReactiveSecurityContextHolder.getContext() 的使用看似正确，但仍会发生此行为。我正在使用 Spring Boot 3.1.8 和 Spring Security 6.1.6。有没有人遇到过类似的问题或者可以深入了解为什么安全上下文显示为空？
我尝试使用 ReactiveSecurityContextHolder.withSecurityContext(Mono.just(securityContext)) 在反应流中设置安全上下文，并期望稍后在同一流中检索此上下文。但是，在设置上下文后立即尝试使用 ReactiveSecurityContextHolder.getContext() 访问上下文时，上下文似乎为空。尽管遵循了 Spring WebFlux 和 Spring Security 反应式应用程序的推荐实践，但这种意外行为还是发生在反应式 Kafka 消费者设置中。
我终于明白了自己的错误；反应式编程对我来说仍然感觉有点违反直觉。该错误是按照我编写安全上下文的顺序发生的。为了使其正常工作并让我的整个流程采用新的安全上下文，我必须将其放置在链的end处，据我所知，这是首先执行的。
我通过遵循此文档发现了这一点，其中他们显示在“handleRequest”方法中，他们将“contextWrite”放在末尾： https://spring.io/blog/2023/03/28/context-propagation-with-project-reactor-1-the-basics
这就是我更正后的代码的样子（值得注意的是，这是一个用于手动测试身份验证的虚拟且简洁的代码）：
private final ReactiveAuthenticationManager authenticationManager;
public <T> Mono<Void> authenticateAndVerifyContext(ReceiverRecord<String, T> record) {
final Mono<SecurityContext> securityContext = initializeSecurityContext(record);
return Mono.just(record)
.then(ReactiveSecurityContextHolder
.getContext()
.flatMap(context -> {
if (context.getAuthentication() != null && context.getAuthentication().isAuthenticated()) {
log.info("Username: {}", context.getAuthentication().getName());
} else {
log.error("Authentication failed!");
}
return Mono.empty();
})
.switchIfEmpty(Mono.defer(() -> {
log.error("Security context is empty!");
return Mono.empty();
})))
.contextWrite(ReactiveSecurityContextHolder.withSecurityContext(securityContext))
.then();
}
public <T> Mono<SecurityContext> initializeSecurityContext(final ReceiverRecord<String, T> record) {
return authenticate(record).map(SecurityContextImpl::new);
}
private <T> Mono<Authentication> authenticate(final ReceiverRecord<String, T> record) {
return extractCredentialsFromBasicAuth(record)
.flatMap(this::authenticate)
.flatMap(this::checkRole);
}
private Mono<Authentication> authenticate(final UsernamePasswordAuthenticationToken authToken) {
return authenticationManager.authenticate(authToken)
.onErrorMap(e -> new BadCredentialsException(
"Authentication failed for user " + authToken.getName()));
}
private <T> Mono<UsernamePasswordAuthenticationToken> extractCredentialsFromBasicAuth(
final ReceiverRecord<String, T> record) {
return Mono.justOrEmpty(record.headers().lastHeader(AUTHORIZATION_KEY))
.map(header -> new String(header.value(), StandardCharsets.UTF_8))
.map(this::decodeBasicAuthHeader)
.flatMap(this::createAuthenticationToken);
}
private String decodeBasicAuthHeader(final String headerValue) {
return new String(Base64.getDecoder().decode(headerValue.substring(6)), StandardCharsets.UTF_8);
}
private Mono<UsernamePasswordAuthenticationToken> createAuthenticationToken(final String credentials) {
final String[] parts = credentials.split(":", 2);
if (parts.length == 2) {
return Mono.just(new UsernamePasswordAuthenticationToken(parts[0], parts[1]));
} else {
return Mono.error(new BadCredentialsException("Invalid basic authentication token"));
}
}
private Mono<Authentication> checkRole(final Authentication authentication) {
return Flux.fromIterable(authentication.getAuthorities())
.map(GrantedAuthority::getAuthority)
.any(role -> REQUIRED_ROLES.stream().anyMatch(requiredRole -> requiredRole.name().equals(role)))
.flatMap(hasRequiredRole -> hasRequiredRole ? Mono.just(authentication) : Mono.empty())
.switchIfEmpty(Mono.error(new AccessDeniedException("Not authorized")));
}