响应式安全：认证、Context与消息授权
Reactor Context能携带身份，但不会证明身份。先读WebFlux请求与资源的固定身份传播探针，再看本页怎样把真正的密码/token验证、请求/方法/对象授权、会话与消息边界接入订阅链。认证与会话、浏览器威胁和OAuth2/OIDC协议仍各有唯一权威正文，本页负责Reactive实现与配置差异。
固定2026-10-26核查：Security7.0.8，commit0102bb2e879e00632f30864785f0f46127877a4d；Boot4.0.9/Framework7.0.10/Reactor3.8.8/Netty1.3.8/RSocket1.1.5/JDK21。完整公开实验使用真实Netty HTTP、真实BCrypt验证、响应式方法代理和RSocket TCP；JWT使用真签名与HTTP JWKS fixture，发布器不是AS。企业LDAP/IdP/KDC及生产TLS未运行。
从凭据转换到认证成功
合成订单服务有alice、bob两个READER和admin。客户端发送Basic凭据，WebFilterChainProxy依次寻找首个匹配的SecurityWebFilterChain；一个较早的宽泛permitAll链同样可能遮住后面的保护。链内AuthenticationWebFilter先按matcher判断是否尝试，再由ServerAuthenticationConverter提取未认证凭据，交ReactiveAuthenticationManager。没有凭据通常继续后续链，由授权/异常翻译决定缺身份响应，不是转换器凭空制造匿名登录。
UserDetailsRepositoryReactiveAuthenticationManager读取ReactiveUserDetailsService，检查账号状态并比较PasswordEncoder；返回已认证Authentication后才构造SecurityContext。它的7.0.8父类默认把密码比较移到boundedElastic，避免BCrypt直接占Netty事件循环；这不保证任意自定义用户查询就是非阻塞。JDBC/LDAP阻塞调用须惰性包装、隔离和限并发；Mono.just(blockingFind())早已执行调用，包Mono没有改变I/O模型。BCrypt cost6只是本机实验加快验证，生产按攻击成本与登录容量测定。
错误密码、未知账号、锁定/禁用等需要在认证层拒绝；日志不打印Authorization、密码或完整token。端点按本例配置返回401；已有身份而能力不够返回403，具体entry point/failure handler可改响应，不把两个数字说成所有协议永远固定。
固定Security7.0.8 AuthenticationWebFilter.java，连续原码，SHA-256为整文件摘要：
public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
return this.requiresAuthenticationMatcher.matches(exchange)
.filter((matchResult) -> matchResult.isMatch())
.flatMap((matchResult) -> this.authenticationConverter.convert(exchange))
.switchIfEmpty(chain.filter(exchange).then(Mono.empty()))
.flatMap((token) -> authenticate(exchange, chain, token))
.onErrorResume(AuthenticationException.class, (ex) -> this.authenticationFailureHandler
.onAuthenticationFailure(new WebFilterExchange(exchange, chain), ex));
}
private Mono<Void> authenticate(ServerWebExchange exchange, WebFilterChain chain, Authentication token) {
return this.authenticationManagerResolver.resolve(exchange)
.flatMap((authenticationManager) -> authenticationManager.authenticate(token))
.switchIfEmpty(Mono
.defer(() -> Mono.error(new IllegalStateException("No provider found for " + token.getClass()))))
.flatMap(
(authentication) -> onAuthenticationSuccess(authentication, new WebFilterExchange(exchange, chain)))
.doOnError(AuthenticationException.class,
(ex) -> logger.debug(LogMessage.format("Authentication failed: %s", ex.getMessage()), ex));
}
protected Mono<Void> onAuthenticationSuccess(Authentication authentication, WebFilterExchange webFilterExchange) {
ServerWebExchange exchange = webFilterExchange.getExchange();
SecurityContextImpl securityContext = new SecurityContextImpl();
securityContext.setAuthentication(authentication);
return this.securityContextRepository.save(exchange, securityContext)
.then(this.authenticationSuccessHandler.onAuthenticationSuccess(webFilterExchange, authentication))
.contextWrite(ReactiveSecurityContextHolder.withSecurityContext(Mono.just(securityContext)));
}成功链先保存到仓库，再执行success handler，并通过contextWrite让后续订阅看到身份；保存在哪里与本次请求传播是两件事。实验API显式NoOp仓库，每次都重新验证；browser链显式WebSession仓库。完整配置与imports可复制如下，不能把片段里的匿名permitAll或合成账号直接部署：
package study.reactivesecurity;
import org.springframework.boot.SpringBootConfiguration;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.security.config.web.server.ServerHttpSecurity;
import org.springframework.security.core.userdetails.MapReactiveUserDetailsService;
import org.springframework.security.core.userdetails.User;
import org.springframework.security.crypto.bcrypt.BCryptPasswordEncoder;
import org.springframework.security.crypto.password.PasswordEncoder;
import org.springframework.security.authentication.UserDetailsRepositoryReactiveAuthenticationManager;
import org.springframework.security.config.Customizer;
import org.springframework.security.web.server.context.NoOpServerSecurityContextRepository;
import org.springframework.security.web.server.SecurityWebFilterChain;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import org.springframework.security.core.context.ReactiveSecurityContextHolder;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.security.config.annotation.method.configuration.EnableReactiveMethodSecurity;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.stereotype.Controller;
import org.springframework.security.config.annotation.rsocket.RSocketSecurity;
import org.springframework.security.rsocket.core.PayloadSocketAcceptorInterceptor;
import org.springframework.core.annotation.Order;
import org.springframework.security.web.server.util.matcher.ServerWebExchangeMatchers;
import org.springframework.security.oauth2.jwt.NimbusReactiveJwtDecoder;
import org.springframework.security.oauth2.jwt.JwtValidators;
import org.springframework.security.oauth2.jwt.JwtClaimValidator;
import org.springframework.security.oauth2.core.DelegatingOAuth2TokenValidator;
import org.springframework.security.oauth2.jose.jws.SignatureAlgorithm;
import org.springframework.security.web.server.context.WebSessionServerSecurityContextRepository;
import org.springframework.security.web.server.savedrequest.NoOpServerRequestCache;
import org.springframework.security.web.server.authentication.HttpStatusServerEntryPoint;
import org.springframework.security.web.server.authentication.logout.HttpStatusReturningServerLogoutSuccessHandler;
import org.springframework.security.web.server.csrf.CsrfToken;
import org.springframework.http.HttpStatus;
import org.springframework.http.HttpMethod;
import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.bind.annotation.PostMapping;
@SpringBootConfiguration
@EnableAutoConfiguration
@Import({SecurityLab.Api.class, SecurityLab.Orders.class, SecurityLab.Messages.class})
@EnableReactiveMethodSecurity
public class SecurityLab {
@Bean(destroyMethod="close") JwtKeys jwtKeys() throws Exception { return new JwtKeys(); }
@Bean @Order(1) SecurityWebFilterChain jwtChain(ServerHttpSecurity http, JwtKeys keys) {
var decoder=NimbusReactiveJwtDecoder.withJwkSetUri(keys.issuer()+"/jwks")
.jwsAlgorithm(SignatureAlgorithm.RS256).build();
decoder.setJwtValidator(new DelegatingOAuth2TokenValidator<>(JwtValidators.createDefaultWithIssuer(keys.issuer()),
new JwtClaimValidator<java.util.List<String>>("aud", aud -> aud!=null&&aud.contains("u31-api"))));
return http.securityMatcher(ServerWebExchangeMatchers.pathMatchers("/jwt/**"))
.securityContextRepository(NoOpServerSecurityContextRepository.getInstance())
.oauth2ResourceServer(oauth -> oauth.jwt(jwt -> jwt.jwtDecoder(decoder)))
.authorizeExchange(a -> a.anyExchange().hasAuthority("SCOPE_read")).build();
}
@Bean PayloadSocketAcceptorInterceptor payloadSecurity(RSocketSecurity socket) {
return socket.simpleAuthentication(Customizer.withDefaults())
.authorizePayload(a -> a.route("admin").hasRole("ADMIN")
.anyRequest().authenticated().anyExchange().permitAll()).build();
}
@Bean PasswordEncoder passwords() { return new BCryptPasswordEncoder(6); }
@Bean MapReactiveUserDetailsService users(PasswordEncoder passwords) {
return new MapReactiveUserDetailsService(
User.withUsername("alice").password(passwords.encode("lab-password")).roles("READER").build(),
User.withUsername("bob").password(passwords.encode("lab-password")).roles("READER").build(),
User.withUsername("admin").password(passwords.encode("lab-password")).roles("READER","ADMIN").build());
}
@Bean UserDetailsRepositoryReactiveAuthenticationManager authenticationManager(
MapReactiveUserDetailsService users, PasswordEncoder passwords) {
var manager=new UserDetailsRepositoryReactiveAuthenticationManager(users);
manager.setPasswordEncoder(passwords);
return manager;
}
@Bean @Order(3) SecurityWebFilterChain apiChain(ServerHttpSecurity http) {
return http.securityContextRepository(NoOpServerSecurityContextRepository.getInstance())
.httpBasic(Customizer.withDefaults())
.authorizeExchange(a -> a.pathMatchers("/api/admin").hasRole("ADMIN")
.anyExchange().authenticated()).build();
}
@Bean @Order(2) SecurityWebFilterChain browserChain(ServerHttpSecurity http) {
var repository=new WebSessionServerSecurityContextRepository();
return http.securityMatcher(ServerWebExchangeMatchers.pathMatchers("/browser/**"))
.securityContextRepository(repository)
.requestCache(cache -> cache.requestCache(NoOpServerRequestCache.getInstance()))
.httpBasic(basic -> basic.securityContextRepository(repository))
.exceptionHandling(e -> e.authenticationEntryPoint(new HttpStatusServerEntryPoint(HttpStatus.UNAUTHORIZED)))
.logout(logout -> logout.requiresLogout(ServerWebExchangeMatchers.pathMatchers(HttpMethod.POST,"/browser/logout"))
.logoutSuccessHandler(new HttpStatusReturningServerLogoutSuccessHandler(HttpStatus.OK)))
.authorizeExchange(a -> a.pathMatchers("/browser/csrf").permitAll().anyExchange().authenticated()).build();
}
@RestController
static class Api {
private final Orders orders;
Api(Orders orders) { this.orders=orders; }
@GetMapping("/api/admin") Mono<String> admin() { return Mono.just("admin-ok"); }
@GetMapping("/api/method") Mono<String> method() { return orders.approve(); }
@GetMapping("/jwt/who") Mono<String> jwt() {
return ReactiveSecurityContextHolder.getContext().map(c -> "jwt:"+c.getAuthentication().getName());
}
@GetMapping("/browser/csrf") Mono<String> csrf(ServerWebExchange exchange) {
Mono<CsrfToken> token=exchange.getAttribute(CsrfToken.class.getName());
return token.map(CsrfToken::getToken);
}
@GetMapping("/browser/who") Mono<String> browserWho() {
return ReactiveSecurityContextHolder.getContext().map(c -> "session:"+c.getAuthentication().getName());
}
@PostMapping("/browser/write") Mono<String> write() { return Mono.just("write-ok"); }
@GetMapping("/api/orders/{owner}") Mono<String> order(@PathVariable String owner) {
return orders.read(owner);
}
@GetMapping("/api/who") Mono<String> who() {
return Mono.just("switch").publishOn(Schedulers.boundedElastic()).flatMap(ignored ->
ReactiveSecurityContextHolder.getContext().map(context -> {
var local=SecurityContextHolder.getContext().getAuthentication();
return "ctx="+context.getAuthentication().getName()+"|threadLocal="+
(local==null?"none":local.getName())+"@"+Thread.currentThread().getName();
}));
}
}
static class Orders {
@PreAuthorize("#owner == authentication.name")
public Mono<String> read(String owner) { return Mono.just("order-of:"+owner); }
@PreAuthorize("hasRole('ADMIN')")
public Mono<String> approve() { return Mono.just("approved"); }
}
@Controller
static class Messages {
@MessageMapping("read") Mono<String> read(String input) {
return ReactiveSecurityContextHolder.getContext().map(c -> "rsocket:"+c.getAuthentication().getName());
}
@MessageMapping("admin") Mono<String> admin(String input) { return Mono.just("rsocket-admin"); }
}
}身份跟随订阅，不跟随任意线程
ReactiveSecurityContextHolder.getContext()返回Mono，从当前订阅Context取Mono<SecurityContext>。SecurityContextServerWebExchangeWebFilter把exchange.getPrincipal适配为该通道，ReactorContextWebFilter负责从配置的repository载入；命令式SecurityContextHolder默认ThreadLocal是另一套载体。7中可选Micrometer上下文桥接也不等于所有线程池自动继承或权限自动刷新。
固定Security7.0.8 ReactiveSecurityContextHolder.java，连续原码，SHA-256为整文件摘要：
/**
* Gets the {@code Mono<SecurityContext>} from Reactor {@link Context}.
* @return the {@code Mono<SecurityContext>}
*/
public static Mono<SecurityContext> getContext() {
// @formatter:off
return Mono.deferContextual(Mono::just)
.cast(Context.class)
.filter(ReactiveSecurityContextHolder::hasSecurityContext)
.flatMap(ReactiveSecurityContextHolder::getSecurityContext);
// @formatter:on
}
private static boolean hasSecurityContext(Context context) {
return context.hasKey(SECURITY_CONTEXT_KEY);
}
private static Mono<SecurityContext> getSecurityContext(Context context) {
return context.<Mono<SecurityContext>>get(SECURITY_CONTEXT_KEY);
}实验/api/who先publishOn(boundedElastic)，再读Reactive Context，返回ctx=alice|threadLocal=none@boundedElastic...。12个alice/bob交错HTTP各自读到自己的身份，之后无凭据仍401。切换线程保留同一订阅上下文；自行subscribe、Future回调或任意Executor开出的新执行，不会因为调用点有身份就自动获得相同上下文。不要把某个线程名当安全判据。
对需要独立订阅的后台工作，先定义有限委托还是执行时重新授权。显式携带身份快照只解决传播，队列等待十分钟后的权限撤销、订单迁租户、委托到期仍需业务裁决。返回框架组合的Mono通常优于在控制器内subscribe；后者还脱离HTTP取消/错误传播。取消本条流会结束本次消费，不能撤销已经提交的订单，也不能把缓存的已认证结果交给另一个用户复用。不要把包含用户身份的Mono放全局cache并跨请求共享。
请求通过后仍需方法与对象授权
请求链按路径约束入口，/api/admin要求ADMIN；/api/method只要求authenticated，再由独立Orders代理的@PreAuthorize检查ADMIN。这是两道实际HTTP对照：alice请求前者403、后者也403；admin两者200。如果请求规则已经拒绝，就不能声称测试证明了方法拦截器。
@EnableReactiveMethodSecurity在7.0.8的useAuthorizationManager默认true；旧资料显式true可以保留，但不能写默认false。被代理方法需返回可适配Publisher，常见Mono/Flux。授权决策与目标调用被组合到订阅中：调用方法拿到Mono并不是副作用已执行。自调用、手工new、绕过代理仍不受该方法advisor保护。后置检查发生在目标执行之后，无法撤回外部副作用；需要先查权限/归属再写，事务与外部副作用仍有各自边界。
固定Security7.0.8 AuthorizationManagerBeforeReactiveMethodInterceptor.java，连续原码，SHA-256为整文件摘要：
Flux<Object> response = preAuthorized(mi,
Flux.defer(() -> adapter.toPublisher(ReactiveMethodInvocationUtils.proceed(mi))));
return KotlinDelegate.asFlow(response);
}
}
if (isMultiValue(type, adapter)) {
Flux<?> result = preAuthorized(mi, Flux.defer(() -> ReactiveMethodInvocationUtils.proceed(mi)));
return (adapter != null) ? adapter.fromPublisher(result) : result;
}
Mono<?> result = preAuthorized(mi, Mono.defer(() -> ReactiveMethodInvocationUtils.proceed(mi)));
return (adapter != null) ? adapter.fromPublisher(result) : result;
}/api/orders/{owner}只是一个明确标注的合成对象标识，服务端方法比较路径owner与已认证principal.name。alice访问bob即使提交?owner=bob仍403；bob访问自己200。真实数据库应把可信tenant/subject与资源ID带入读/更新条件，不能把这个简化owner路径当生产表设计；issuer/tenant命名空间与TOCTOU的权威解释见资源授权。Reactive表达式可调用返回Mono<Boolean>的受控bean，查询仍需有界，不在表达式里block等数据库。
WebSession、退出与并发登录
API/JWT链使用NoOpServerSecurityContextRepository，成功仅供当前请求；它不等于自动禁用所有Cookie，也不等于可关闭CSRF。Basic凭据可能被浏览器自动附带，威胁取决于凭据运输。browser链为展示会话保存，给全链和Basic filter都配置同一个WebSessionServerSecurityContextRepository；仓库保存身份时改变session id。只设置一个全局仓库后猜测所有认证filter都以相同方式持久化，容易误判默认值。
实验先GET/browser/csrf创建会话，再携凭据GET who；id改变、后继只带cookie得到session:alice。POST缺/错CSRF403，正确token200；POST logout后cookie请求401，再验证凭据恢复。默认XOR CSRF request handler暴露的是可提交token表示，测试从当前会话端点取回，不硬编码旧token。本例默认logout清除SecurityContext与CSRF并更换session id；它不销毁整个WebSession，普通业务属性可能仍保留，不会自动撤AS access/refresh或企业IdP session。若业务要求整会话销毁，应在保留默认logout handlers的基础上追加WebSessionServerLogoutHandler，并验证业务属性、旧cookie/CSRF及重新登录。
固定Security7.0.8 WebSessionServerSecurityContextRepository.java，连续原码，SHA-256为整文件摘要：
public Mono<Void> save(ServerWebExchange exchange, @Nullable SecurityContext context) {
return exchange.getSession().doOnNext((session) -> {
if (context == null) {
session.getAttributes().remove(this.springSecurityContextAttrName);
logger.debug(LogMessage.format("Removed SecurityContext stored in WebSession: '%s'", session));
}
else {
session.getAttributes().put(this.springSecurityContextAttrName, context);
logger.debug(LogMessage.format("Saved SecurityContext '%s' in WebSession: '%s'", context, session));
}
}).flatMap(WebSession::changeSessionId);
}
@Override
public Mono<SecurityContext> load(ServerWebExchange exchange) {
Mono<SecurityContext> result = exchange.getSession().flatMap((session) -> {并发会话为具体配置教学，未运行。依赖本lab的web/config；在已配置formLogin/oauth2Login/Basic的ServerHttpSecurity加：
import org.springframework.security.core.session.InMemoryReactiveSessionRegistry;
import org.springframework.security.core.session.ReactiveSessionRegistry;
import org.springframework.security.web.server.authentication.SessionLimit;
@Bean ReactiveSessionRegistry reactiveSessionRegistry() {
return new InMemoryReactiveSessionRegistry();
}
// 在已有http链build前配置：
http.sessionManagement(s -> s.concurrentSessions(c -> c.maximumSessions(SessionLimit.of(1))));默认不限并发；显式limit通过成功handler登记、检查。默认超限淘汰较少近期使用的会话；若希望拒绝新登录，配置PreventLoginServerMaximumSessionsExceededHandler，不要把淘汰与阻止新会话当成一回事。SessionLimit可按Authentication异步返回数量，Mono.empty表示不设限制。自定义success handler可能改变默认串联，须保留登记/检查所需handler。InMemoryReactiveSessionRegistry只认识单进程；共享WebSession存储不自动把注册表变成集群一致，失效需同时清注册与SessionStore记录并考虑竞态。本地并发控制不知道IdP会话，要另实现相应退出/撤销。
Reactive Resource Server的验证入口
协议规则见OAuth正文，Reactive实现为AuthenticationWebFilter、ServerBearerTokenAuthenticationConverter、JwtReactiveAuthenticationManager及ReactiveJwtDecoder；不要注入Servlet的JwtDecoder/Filter来替代。实验JWT链优先匹配/jwt/**，显式RS256、可信issuer/JWKS、默认validator加aud=u31-api，权限要求SCOPE_read。真RSA签名与HTTP JWKS取钥后，错签名/issuer/audience/过期/未来nbf均401，每个反例后valid200；scope不足403。
默认时间校验有60秒skew，默认允许缺exp/nbf，本lab并未像U30额外强制exp或改零skew，反例取相差120秒。显式替换validator会丢默认约束，需组合保留。JWK缓存、撤销和algorithm builder差异仍以U30固定实现为依据；Reactive test工具mockJwt不能证明这里发生过验签/取钥。
package study.reactivesecurity;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.util.Date;
import java.util.List;
import com.sun.net.httpserver.HttpServer;
import com.nimbusds.jose.JOSEException;
import com.nimbusds.jose.JWSAlgorithm;
import com.nimbusds.jose.JWSHeader;
import com.nimbusds.jose.crypto.RSASSASigner;
import com.nimbusds.jose.jwk.JWKSet;
import com.nimbusds.jose.jwk.RSAKey;
import com.nimbusds.jose.jwk.gen.RSAKeyGenerator;
import com.nimbusds.jwt.JWTClaimsSet;
import com.nimbusds.jwt.SignedJWT;
/** 本机真实签名和JWKS网络fixture；不模拟已认证SecurityContext，不是授权服务器。 */
public final class JwtKeys implements AutoCloseable {
private final RSAKey key;
private final HttpServer server;
public JwtKeys() throws Exception {
key=new RSAKeyGenerator(2048).keyID("u31-key").generate();
server=HttpServer.create(new InetSocketAddress("127.0.0.1",0),0);
server.createContext("/jwks",exchange->{
var bytes=new JWKSet(key.toPublicJWK()).toString().getBytes(StandardCharsets.UTF_8);
exchange.getResponseHeaders().set("Content-Type","application/json");
exchange.sendResponseHeaders(200,bytes.length);
try(var output=exchange.getResponseBody()){output.write(bytes);}
});
server.start();
}
public String issuer() { return "http://127.0.0.1:"+server.getAddress().getPort(); }
public String token(String variant) throws JOSEException {
var claims=new JWTClaimsSet.Builder().issuer(variant.equals("issuer")?"https://untrusted.invalid":issuer())
.audience(variant.equals("audience")?"other-api":"u31-api").subject("alice")
.expirationTime(Date.from(Instant.now().plusSeconds(variant.equals("expired")?-120:120)))
.notBeforeTime(Date.from(Instant.now().plusSeconds(variant.equals("future")?120:-10)))
.claim("scope",variant.equals("scope")?"other":"read").claim("tenant","t1").build();
var jwt=new SignedJWT(new JWSHeader.Builder(JWSAlgorithm.RS256).keyID("u31-key").build(),claims);
RSAKey signer=variant.equals("signature")?new RSAKeyGenerator(2048).generate():key;
jwt.sign(new RSASSASigner(signer));
return jwt.serialize();
}
@Override public void close(){server.stop(0);System.out.println("U31OBSERVE jwks fixtureStopped=true");}
}Opaque的Reactive适配为配置教学，未运行，依赖resource-server与受信AS：
import org.springframework.security.oauth2.server.resource.introspection.SpringReactiveOpaqueTokenIntrospector;
import org.springframework.security.oauth2.server.resource.introspection.ReactiveOpaqueTokenIntrospector;
@Bean ReactiveOpaqueTokenIntrospector introspector() {
return SpringReactiveOpaqueTokenIntrospector.withIntrospectionUri("https://as.example/introspect")
.clientId("orders-rs").clientSecret(System.getenv("INTROSPECTION_SECRET")).build();
}
// 一条独立的opaque资源链，不与本lab JWT spec叠在一起：
http.oauth2ResourceServer(o -> o.opaqueToken(Customizer.withDefaults()));真实部署secret来自受控配置，AS地址固定受信；检查active之外的aud/tenant策略需要装饰introspector，不能以远程active替代资源归属。连接/响应预算、429/5xx重试和fail closed需明示：认证服务故障不能变成功身份；业务可按显式failure handler区分503与无效凭据401，不能宣称框架默认一律503。本地JWT验证与introspection对撤销的依赖不同，协议实跑见U30。
多issuer使用JwtIssuerReactiveAuthenticationManagerResolver.fromTrustedIssuers(...)并配置resource-server的authenticationManagerResolver；未验签iss仅作预先信任集合内的路由提示，每个manager仍验签/iss/aud。动态注册新issuer需可信控制面，禁止任意URL拉metadata/JWK；business tenant来自可信成员关系。Bearer默认从Authorization读取，不为便利开启query token，避免URL日志/缓存泄露。下游传播优先使用绑定authorized client的OAuth WebClient filter，不能给任意目标自动附入站token。
Reactive OAuth Login与出站Client
以下为独立配置教学，未接真实OP运行。增加spring-boot-starter-oauth2-client，应用是WebFlux，配置Registration/provider和外部基址；issuer必须为受信OP，redirect与OP登记严格一致：
spring:
security:
oauth2:
client:
registration:
orders:
provider: company
client-id: orders-web
client-secret: ${OIDC_SECRET}
authorization-grant-type: authorization_code
scope: openid,profile
redirect-uri: '{baseUrl}/login/oauth2/code/{registrationId}'
provider:
company:
issuer-uri: https://id.exampleimport org.springframework.security.oauth2.client.registration.ReactiveClientRegistrationRepository;
import org.springframework.security.oauth2.client.web.server.ServerOAuth2AuthorizedClientRepository;
import org.springframework.security.oauth2.client.web.DefaultReactiveOAuth2AuthorizedClientManager;
import org.springframework.security.oauth2.client.ReactiveOAuth2AuthorizedClientProviderBuilder;
import org.springframework.security.oauth2.client.web.reactive.function.client.ServerOAuth2AuthorizedClientExchangeFilterFunction;
import org.springframework.web.reactive.function.client.WebClient;
@Bean WebClient ordersWebClient(ReactiveClientRegistrationRepository registrations,
ServerOAuth2AuthorizedClientRepository clients) {
var manager=new DefaultReactiveOAuth2AuthorizedClientManager(registrations,clients);
manager.setAuthorizedClientProvider(ReactiveOAuth2AuthorizedClientProviderBuilder.builder()
.authorizationCode().refreshToken().clientCredentials().build());
var filter=new ServerOAuth2AuthorizedClientExchangeFilterFunction(manager);
return WebClient.builder().baseUrl("https://orders.example").filter(filter).build();
}
// 在独立浏览器链启用：
http.oauth2Login(Customizer.withDefaults()).oauth2Client(Customizer.withDefaults());
// 每次调用选明确registration，不对所有WebClient请求默认附token：
ordersWebClient.get().uri("/orders").attributes(
ServerOAuth2AuthorizedClientExchangeFilterFunction.clientRegistrationId("orders"))
.retrieve().bodyToMono(String.class);ReactiveRegistration仓库管理provider/client配置，authorized-client仓库/服务保存principal+registration的access/refresh状态。DefaultReactiveOAuth2AuthorizedClientManager有ServerWebExchange语境；后台任务用AuthorizedClientServiceReactiveOAuth2AuthorizedClientManager和ReactiveOAuth2AuthorizedClientService，明确机器还是用户授权。authorization_code需浏览器授权/回调，refresh需已有refresh token，client_credentials代表机器。JWT bearer与token exchange使用所选Reactive provider并提供相应assertion/subject-token resolver，不能靠加scope凭空获得；7已移除password grant，不给旧密码模式复制入口。Client secret/basic/post/private_key_jwt/mTLS须按AS注册与对应token WebClient配置匹配，PKCE不能替代客户端认证，传输证书信任也不等于用户授权。
如AS明确支持JWT bearer或token exchange，完整manager还需加入对应provider。以下替换上面manager的provider装配，registration的grant type分别设urn:ietf:params:oauth:grant-type:jwt-bearer或urn:ietf:params:oauth:grant-type:token-exchange，并将真正经过验证的JwtAuthenticationToken作为授权上下文principal；不是把浏览器参数字符串当可信assertion：
import org.springframework.security.oauth2.client.DelegatingReactiveOAuth2AuthorizedClientProvider;
import org.springframework.security.oauth2.client.JwtBearerReactiveOAuth2AuthorizedClientProvider;
import org.springframework.security.oauth2.client.TokenExchangeReactiveOAuth2AuthorizedClientProvider;
var ordinary=ReactiveOAuth2AuthorizedClientProviderBuilder.builder()
.authorizationCode().refreshToken().clientCredentials().build();
var assertion=new JwtBearerReactiveOAuth2AuthorizedClientProvider();
var exchange=new TokenExchangeReactiveOAuth2AuthorizedClientProvider();
manager.setAuthorizedClientProvider(new DelegatingReactiveOAuth2AuthorizedClientProvider(
ordinary,assertion,exchange));默认assertion/subject resolver从context.getPrincipal().getPrincipal()取Jwt/OAuth2Token；另有受控来源才设置setJwtAssertionResolver/setSubjectTokenResolver，需要actor时另设setActorTokenResolver。AS须校验subject/actor、目标audience和允许委托范围。原token受众错误、来源不可信或权限提升应拒绝，恢复是修受控注册/授权并重新取得合格subject，不是关掉验证。该配置未与真实AS运行。
ServerAuthorizationRequestRepository<OAuth2AuthorizationRequest>保存授权请求，默认实现WebSessionOAuth2ServerAuthorizationRequestRepository使用WebSession；resolver改参数仍要保留state/nonce/PKCE绑定。OidcReactiveOAuth2UserService消费ID/UserInfo并映射本地权限；ID token不拿来当RS access。token endpoint WebClient设置连接/读预算，失败时保留可恢复关联或重新发起授权，别重放已消费code。refresh invalid_grant需清失效authorized client并重新授权，固定filter/manager的failure handler需核实接线，不能假定任意手工manager自动清理所有存储。
退出分三层：本地logout清安全身份/CSRF，整WebSession销毁需额外handler；RP initiated可配置OidcClientInitiatedServerLogoutSuccessHandler(registrations)及合法post_logout_redirect_uri；back-channel需在保留oauth2Login的浏览器链显式启用：
http.oidcLogout(o -> o.backChannel(Customizer.withDefaults()));实际处理为OidcBackChannelLogoutWebFilter接收请求，OidcBackChannelLogoutReactiveAuthenticationManager通过ReactiveJwtDecoder验签及claims，再由OidcBackChannelServerLogoutHandler与ReactiveOidcSessionRegistry定位iss+sid/sub相关session并执行退出。registry存关联，不负责代替decoder验token。7默认registry可内存，需要与cookie名称（该handler实际默认SESSION）、WebSession持久化和多节点映射一致。OP outage/错issuer/aud/签名不得失效不相关会话；新登录与旧session清理有竞态，IdP退出不自动证明资源token全部撤销。固定源码配置细节与正常/故障预期应在接入具体OP时逐项复验，本票没运行这些退出路径。
RSocket在每条业务请求上裁决
RSocket走SecuritySocketAcceptorInterceptor→PayloadSocketAcceptorInterceptor→认证/授权PayloadInterceptor，不能用HTTP路径过滤器保护route。实际工程在RSocketServer显式安装socket acceptor interceptor；只声明bean而没有接到transport，客户端成功不证明安全链生效。
本lab允许SETUP，anyRequest().authenticated()保护业务请求，admin route另要求ADMIN。客户端通过SimpleAuthenticationEncoder把UsernamePasswordMetadata编码到message/x.rsocket.authentication.v0，服务端simpleAuthentication调用真实密码manager；这不同于HTTP Authorization: Basic。无/错凭据失败后，alice读成功；alice admin失败、admin成功；同连接bob读到bob，随后匿名仍拒绝，再alice恢复。异常是协议ApplicationErrorException，不用HTTP401/403给RSocket帧命名。
package study.reactivesecurity;
import java.time.Duration;
import org.junit.jupiter.api.Test;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.messaging.rsocket.RSocketStrategies;
import org.springframework.security.rsocket.core.SecuritySocketAcceptorInterceptor;
import org.springframework.security.rsocket.metadata.UsernamePasswordMetadata;
import org.springframework.security.rsocket.metadata.SimpleAuthenticationEncoder;
import org.springframework.util.MimeTypeUtils;
import io.rsocket.core.RSocketServer;
import io.rsocket.transport.netty.server.TcpServerTransport;
import io.rsocket.transport.netty.server.CloseableChannel;
import static org.junit.jupiter.api.Assertions.*;
@DirtiesContext(classMode=DirtiesContext.ClassMode.AFTER_CLASS)
@SpringBootTest(classes=SecurityLab.class,webEnvironment=SpringBootTest.WebEnvironment.RANDOM_PORT,
properties={"server.address=127.0.0.1","spring.main.web-application-type=reactive"})
class RSocketAuthenticationTest {
@Autowired RSocketMessageHandler handler;
@Autowired SecuritySocketAcceptorInterceptor security;
@Test void requestsAuthenticateAndAuthorizeIndependentlyOnOneRealConnection() {
var strategies=RSocketStrategies.builder().encoder(new SimpleAuthenticationEncoder()).build();
CloseableChannel listener=RSocketServer.create(handler.responder())
.interceptors(registry->registry.forSocketAcceptor(security))
.bindNow(TcpServerTransport.create("127.0.0.1",0));
var requester=RSocketRequester.builder().rsocketStrategies(strategies).tcp("127.0.0.1",listener.address().getPort());
try {
assertThrows(io.rsocket.exceptions.ApplicationErrorException.class,()->call(requester,"read",null,null));
assertThrows(io.rsocket.exceptions.ApplicationErrorException.class,()->call(requester,"read","alice","wrong"));
String value=call(requester,"read","alice","lab-password");
assertEquals("rsocket:alice",value);
System.out.println("U31OBSERVE rsocket validUser=alice value="+value);
assertThrows(io.rsocket.exceptions.ApplicationErrorException.class,()->call(requester,"admin","alice","lab-password"));
assertEquals("rsocket-admin",call(requester,"admin","admin","lab-password"));
assertEquals("rsocket:bob",call(requester,"read","bob","lab-password"));
assertThrows(io.rsocket.exceptions.ApplicationErrorException.class,()->call(requester,"read",null,null));
assertEquals("rsocket:alice",call(requester,"read","alice","lab-password"));
System.out.println("U31OBSERVE rsocket missingAndWrongDenied=true roleDenied=true adminAllowed=true perRequestIdentity=true recovered=true");
} finally {
requester.dispose();listener.dispose();listener.onClose().block(Duration.ofSeconds(5));
System.out.println("U31OBSERVE rsocket listenerDisposed="+listener.isDisposed());
}
}
static String call(RSocketRequester requester,String route,String user,String password) {
var request=requester.route(route);
if(user!=null) request.metadata(new UsernamePasswordMetadata(user,password),
MimeTypeUtils.parseMimeType("message/x.rsocket.authentication.v0"));
return request.data("u31").retrieveMono(String.class).block(Duration.ofSeconds(5));
}
}如果一连接专属单个用户，可在SETUP认证并复用连接身份；如果连接池供多用户使用，必须逐请求携带身份或明示有限委托，不能把第一个用户身份当整条连接授权。可同时要求SETUP机器身份和请求用户权限，但要分别设计认证器与authority范围，不把人类用户凭据当连池机器票。JWT payload改用BearerTokenMetadata和对应encoder/decoder、RSocketSecurity.jwt；可信issuer/时间/aud策略仍要显式设置。流启动后撤权不会自动回收全部已产生数据，需定义每次请求/每段流的再检查与取消规则。SETUP双层、JWT payload、TLS、续传与集群撤销本lab未运行。
浏览器防护、扩展与验证边界
CSRF、CORS、Headers的威胁模型回指U29。Reactive是CsrfWebFilter/ServerCsrfTokenRepository、CorsWebFilter/ServerHttpSecurity.cors、ServerHttpHeadersWriter等实现，不能拷Servlet CookieCsrfTokenRepository/Filter实例。预检无登录cookie，应在身份裁决前正确处理CORS；允许origin+credentials需严格白名单，CORS不替代服务端认证。CSRF token是延迟Mono，只有订阅对应属性才实际创建/写入；不要在独立subscribe中生成，或将受害者token返回允许任意origin读取。
一个独立浏览器部署可使用下列完整Bean约束跨域和HTTP方法，并在其ServerHttpSecurity链中启用相应source/headers；本lab只保留默认防护，这个额外配置未运行，origin应替换为受控部署值：
import java.util.List;
import org.springframework.context.annotation.Bean;
import org.springframework.http.HttpMethod;
import org.springframework.web.cors.CorsConfiguration;
import org.springframework.web.cors.reactive.CorsConfigurationSource;
import org.springframework.web.cors.reactive.UrlBasedCorsConfigurationSource;
import org.springframework.security.web.server.firewall.ServerWebExchangeFirewall;
import org.springframework.security.web.server.firewall.StrictServerWebExchangeFirewall;
@Bean CorsConfigurationSource corsSource() {
var cors=new CorsConfiguration();
cors.setAllowedOrigins(List.of("https://console.example"));
cors.setAllowedMethods(List.of("GET","POST","OPTIONS"));
cors.setAllowedHeaders(List.of("Content-Type","X-CSRF-TOKEN"));
cors.setAllowCredentials(true);
cors.setMaxAge(600L); // 秒；浏览器缓存预检，不是授权TTL
var source=new UrlBasedCorsConfigurationSource();
source.registerCorsConfiguration("/**",cors);
return source;
}
@Bean ServerWebExchangeFirewall firewall() {
var firewall=new StrictServerWebExchangeFirewall();
firewall.setAllowedHttpMethods(List.of(HttpMethod.GET,HttpMethod.POST,HttpMethod.OPTIONS));
return firewall;
}
// 在使用上面Bean的同一应用的浏览器链build前：
http.cors(c -> c.configurationSource(corsSource()))
.headers(h -> h.contentSecurityPolicy(c -> c.policyDirectives("default-src 'self'; frame-ancestors 'none'")));@EnableWebFluxSecurity配置会把ServerWebExchangeFirewall Bean接到WebFilterChainProxy；不要再注册第二个同类过滤链代理。PUT确有业务需求才显式扩方法；预期恶意origin无可读凭据响应，合法预检/带CSRF的POST正常，修复前后都验证。CSP需要根据实际脚本/样式来源调整，不能以'self'样例宣称任意站点已兼容。
HTTP严格校验由StrictServerWebExchangeFirewall保护方法、路径/headers等；编码分隔符、异常路径是否拒绝按固定版本/反向代理解码层核对。允许某个编码形式意味着扩大可达路径，应同时复查匹配/路由与资源规则，不能为消除400/401盲目全放行。HTTPS及HSTS依赖实际安全请求/可信转发，应用不得相信任意客户端X-Forwarded-Proto。默认头不是CSP业务策略，WebSocket/RSocket也有各自消息边界。上述完整HTTP攻击矩阵未在本lab重复运行。
Reactive X509配置入口为ServerHttpSecurity.x509，principalExtractor和ReactiveAuthenticationManager各有角色，证书链验证先由TLS层完成。默认7.0.8 extractor是SubjectX500PrincipalExtractor；Servlet/Reactive因子追加行为不同，CN字符串不是证书有效性证据。LDAP/SAML/WebAuthn/CAS/Kerberos通常是Servlet或阻塞API，不能猜ServerHttpSecurity都有同名DSL；完整配置与不支持边界见企业认证。Reactive OTT在7.0.8存在，但@EnableMultiFactorAuthentication明确暂不支持Reactive，不能靠静态FACTOR字符串宣称完成MFA。
观测应分别记录链选择、认证失败类型、授权拒绝位置与依赖超时；认证通过指标不等于业务成功。启用ObservationRegistry可观测WebFilter/认证/授权，升级时检查额外包裹与Context传播，标签使用有限registration/route/结果类别，不拿username、原始URI、token作高基数/敏感标签。该集成配置解释不是本次生产指标收集/压测。
WebTestClient.bindToApplicationContext适合查过滤器/响应布线，没有真实socket；mockUser/mockJwt/mockOidcLogin/mockX509、@WithMockUser等替身适合授权规则，但跳过密码、JWK网络、TLS链与OP协议。当前公开测试由JDK HttpClient向真实Netty发请求，RSocket用真实TCP；不要把测试框架中认证替身的成功换算为认证器通过。方法测试还须从Spring代理调用并订阅Publisher，避免仅断言得到一个非空Mono。
容量、撤销窗口与恢复责任
模型：稳定200次登录/秒，密码比较平均50ms，约10个同时进行的密码任务；若资源只容纳4个CPU任务，切到boundedElastic不会创造算力。上限不仅是线程数，还包含密码cost、队列等待、用户目录连接、HTTP时限。可选择专用有界密码scheduler配登录接纳，或复用受控隔离池配严格限流；前者隔离强但需独立调参，后者共享资源简单却可能被其他阻塞依赖拖垮。测得服务时间与尾延迟后再配置，不拿此算式当本机吞吐。
另一个场景：1万条RSocket连接、平均每条20条请求/秒，目标撤权10秒。仅SETUP身份会无限长保留旧权限，直到断线或另外失效；逐请求校验20万次/秒会放大权威服务负载。候选一每请求查权威/短缓存并失败关闭，成本高、撤销窗口可预算；候选二短期委托凭据+版本/失效通知+周期检查，低延迟但必须把TTL、缓存、广播/扫描和故障检测分配进10秒。长流已发送内容不可收回，敏感输出要分段检查。窗口责任由身份平台、应用session/连接管理与资源服务分别承担，不能让客户端“关闭按钮”作为失效协议。
故障恢复依次确认：到底选了哪条链/route → 输入有没有转换成凭据 → manager有没有查询/验密/验签 → Context从哪条订阅来 → 方法是否过代理 → 对象归属是否含可信主体 → session/token/连接怎样撤销。恢复测试保留合法控制和匿名后继，JWT坏claim后再发valid、RSocket拒绝后同连接再发合法、logout后再凭据登录。企业服务不可用不降为匿名成功，也不通过取消验证/扩大issuer名单“恢复”。
官方来源与复现结果
固定Reactive官方目录含认证/请求与方法授权/OAuth/防护/集成/测试；RSocket、并发会话的配置与失败边界如上。固定nav没有列全Reactive OTT，范围审计补入，不以目录标题数证明覆盖完整。
公开实验和其中运行JSON记录真实输入、JDK/OS/依赖、命令/退出、源码hash与资源清理；源码核查/配置教学、受控JWKS fixture、真实HTTP/TCP分别计。无企业IdP运行、TLS、自然TTL等待、生产容量或多节点一致性证据。教材验收不表示学习者已经掌握。
闭卷验收
本页 20 道闭卷题收进了下方自测面板：先回忆，再展开参考答案与评分点，最后给三档自评。