From 648f8f999bfa79627d6b49c54e533f9c0803bccf Mon Sep 17 00:00:00 2001 From: Spencer Gibb Date: Mon, 16 Dec 2019 19:28:43 -0500 Subject: [PATCH] Propagates reactor context thru HystrixCommand. Because of the RxJava and the Reactive Streams integration, subsequent filters lost the reactor context. The passes the context back to the filter.chain() call. Fixes gh-1258 --- .../factory/HystrixGatewayFilterFactory.java | 79 +++++---- ...rixGatewayFilterFactoryPrincipalTests.java | 154 ++++++++++++++++++ .../gateway/test/BaseWebClientTests.java | 26 ++- 3 files changed, 214 insertions(+), 45 deletions(-) create mode 100644 spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/filter/factory/HystrixGatewayFilterFactoryPrincipalTests.java diff --git a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/HystrixGatewayFilterFactory.java b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/HystrixGatewayFilterFactory.java index 1b3a6104..be4f01a3 100644 --- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/HystrixGatewayFilterFactory.java +++ b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/HystrixGatewayFilterFactory.java @@ -27,6 +27,7 @@ import com.netflix.hystrix.HystrixObservableCommand; import com.netflix.hystrix.HystrixObservableCommand.Setter; import com.netflix.hystrix.exception.HystrixRuntimeException; import reactor.core.publisher.Mono; +import reactor.util.context.Context; import rx.Observable; import rx.RxReactiveStreams; import rx.Subscription; @@ -119,45 +120,47 @@ public class HystrixGatewayFilterFactory @Override public Mono filter(ServerWebExchange exchange, GatewayFilterChain chain) { - RouteHystrixCommand command = new RouteHystrixCommand(config.setter, - config.fallbackUri, exchange, chain); + return Mono.deferWithContext(context -> { + RouteHystrixCommand command = new RouteHystrixCommand(config.setter, + config.fallbackUri, exchange, chain, context); - return Mono.create(s -> { - Subscription sub = command.toObservable().subscribe(s::success, - s::error, s::success); - s.onCancel(sub::unsubscribe); - }).onErrorResume((Function>) throwable -> { - if (throwable instanceof HystrixRuntimeException) { - HystrixRuntimeException e = (HystrixRuntimeException) throwable; - HystrixRuntimeException.FailureType failureType = e - .getFailureType(); + return Mono.create(s -> { + Subscription sub = command.toObservable().subscribe(s::success, + s::error, s::success); + s.onCancel(sub::unsubscribe); + }).onErrorResume((Function>) throwable -> { + if (throwable instanceof HystrixRuntimeException) { + HystrixRuntimeException e = (HystrixRuntimeException) throwable; + HystrixRuntimeException.FailureType failureType = e + .getFailureType(); - switch (failureType) { - case TIMEOUT: - return Mono.error(new TimeoutException()); - case SHORTCIRCUIT: - return Mono.error(new ServiceUnavailableException()); - case COMMAND_EXCEPTION: { - Throwable cause = e.getCause(); + switch (failureType) { + case TIMEOUT: + return Mono.error(new TimeoutException()); + case SHORTCIRCUIT: + return Mono.error(new ServiceUnavailableException()); + case COMMAND_EXCEPTION: { + Throwable cause = e.getCause(); - /* - * We forsake here the null check for cause as - * HystrixRuntimeException will always have a cause if the - * failure type is COMMAND_EXCEPTION. - */ - if (cause instanceof ResponseStatusException - || AnnotatedElementUtils.findMergedAnnotation( - cause.getClass(), - ResponseStatus.class) != null) { - return Mono.error(cause); + /* + * We forsake here the null check for cause as + * HystrixRuntimeException will always have a cause if the + * failure type is COMMAND_EXCEPTION. + */ + if (cause instanceof ResponseStatusException + || AnnotatedElementUtils.findMergedAnnotation( + cause.getClass(), + ResponseStatus.class) != null) { + return Mono.error(cause); + } + } + default: + break; } } - default: - break; - } - } - return Mono.error(throwable); - }).then(); + return Mono.error(throwable); + }).then(); + }); } @Override @@ -222,17 +225,21 @@ public class HystrixGatewayFilterFactory private final GatewayFilterChain chain; + private final Context context; + RouteHystrixCommand(Setter setter, URI fallbackUri, ServerWebExchange exchange, - GatewayFilterChain chain) { + GatewayFilterChain chain, Context context) { super(setter); this.fallbackUri = fallbackUri; this.exchange = exchange; this.chain = chain; + this.context = context; } @Override protected Observable construct() { - return RxReactiveStreams.toObservable(this.chain.filter(exchange)); + return RxReactiveStreams + .toObservable(this.chain.filter(exchange).subscriberContext(context)); } @Override diff --git a/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/filter/factory/HystrixGatewayFilterFactoryPrincipalTests.java b/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/filter/factory/HystrixGatewayFilterFactoryPrincipalTests.java new file mode 100644 index 00000000..2d808de7 --- /dev/null +++ b/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/filter/factory/HystrixGatewayFilterFactoryPrincipalTests.java @@ -0,0 +1,154 @@ +/* + * Copyright 2013-2019 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.gateway.filter.factory; + +import java.security.Principal; +import java.util.Collections; +import java.util.Map; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.junit.Test; +import org.junit.runner.RunWith; +import reactor.core.publisher.Mono; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.SpringBootConfiguration; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.cloud.gateway.filter.GatewayFilter; +import org.springframework.cloud.gateway.filter.ratelimit.KeyResolver; +import org.springframework.cloud.gateway.filter.ratelimit.PrincipalNameKeyResolver; +import org.springframework.cloud.gateway.route.RouteLocator; +import org.springframework.cloud.gateway.route.builder.RouteLocatorBuilder; +import org.springframework.cloud.gateway.test.BaseWebClientTests; +import org.springframework.cloud.netflix.ribbon.RibbonClient; +import org.springframework.cloud.netflix.ribbon.RibbonClients; +import org.springframework.context.annotation.Bean; +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.core.userdetails.UserDetails; +import org.springframework.security.web.server.SecurityWebFilterChain; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.springframework.boot.test.context.SpringBootTest.WebEnvironment.RANDOM_PORT; + +@RunWith(SpringRunner.class) +@SpringBootTest(webEnvironment = RANDOM_PORT, properties = "debug=true") +@DirtiesContext +public class HystrixGatewayFilterFactoryPrincipalTests extends BaseWebClientTests { + + @Autowired + private TestPrincipalGatewayFilterFactory testFilterFactory; + + @Test + public void hystrixPrincipalNotLost() { + testClient.get().uri("/hystrixprincipal").headers(httpHeaders -> { + httpHeaders.setBasicAuth("user", "password"); + httpHeaders.set("Host", "www.hystrixsecurity.org"); + }).exchange().expectStatus().isOk().expectBody().jsonPath("$.principal") + .isEqualTo("user"); + assertThat(testFilterFactory.resolvedPrincipal).isEqualTo("user"); + } + + @RestController + @SpringBootConfiguration + @EnableAutoConfiguration + @RibbonClients({ + @RibbonClient(name = "testservice", configuration = TestRibbonConfig.class) }) + public static class TestConfig { + + @Value("${test.uri}") + private String uri; + + @RequestMapping("/httpbin/hystrixprincipal") + public Mono> hystrixPrincipal(Mono principal) { + return principal.map(Principal::getName).defaultIfEmpty("Unknown") + .map(s -> Collections.singletonMap("principal", s)); + } + + @Bean + public RouteLocator hystrixRouteLocator(RouteLocatorBuilder builder, + TestPrincipalGatewayFilterFactory filterFactory) { + return builder.routes() + .route("hystrix_security", r -> r.host("**.hystrixsecurity.org") + .filters(f -> f.prefixPath("/httpbin") + .hystrix(config -> config.setName("securitycmd")) + .filter(filterFactory.apply(""))) + .uri(uri)) + .build(); + } + + @Bean + public TestPrincipalGatewayFilterFactory testPrincipalGatewayFilterFactory() { + return new TestPrincipalGatewayFilterFactory(); + } + + @Bean + public RecursiveHttpbinFilter recursiveHttpbinFilter() { + return new RecursiveHttpbinFilter(); + } + + @Bean + SecurityWebFilterChain springWebFilterChain(ServerHttpSecurity http) { + return http.httpBasic().and().authorizeExchange() + .pathMatchers("/hystrixprincipal").authenticated().anyExchange() + .permitAll().and().build(); + } + + @Bean + @SuppressWarnings("deprecation") + public MapReactiveUserDetailsService reactiveUserDetailsService() { + UserDetails user = User.withDefaultPasswordEncoder().username("user") + .password("password").roles("USER").build(); + return new MapReactiveUserDetailsService(user); + } + + } + + public static class TestPrincipalGatewayFilterFactory + extends AbstractGatewayFilterFactory { + + private final Log log = LogFactory + .getLog(TestPrincipalGatewayFilterFactory.class); + + private KeyResolver keyResolver = new PrincipalNameKeyResolver(); + + private String resolvedPrincipal; + + public TestPrincipalGatewayFilterFactory() { + super(Object.class); + } + + @Override + public GatewayFilter apply(Object config) { + return (exchange, chain) -> keyResolver.resolve(exchange) + .defaultIfEmpty("Empty Principal").flatMap(name -> { + resolvedPrincipal = name; + return chain.filter(exchange); + }); + } + + } + +} diff --git a/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/BaseWebClientTests.java b/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/BaseWebClientTests.java index 7a25b9e7..aab9433a 100644 --- a/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/BaseWebClientTests.java +++ b/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/BaseWebClientTests.java @@ -28,6 +28,7 @@ import reactor.core.publisher.Mono; import org.springframework.boot.SpringBootConfiguration; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.web.server.LocalServerPort; +import org.springframework.cloud.gateway.filter.GatewayFilterChain; import org.springframework.cloud.gateway.filter.GlobalFilter; import org.springframework.cloud.gateway.route.Route; import org.springframework.cloud.netflix.ribbon.RibbonClient; @@ -41,6 +42,7 @@ import org.springframework.http.client.reactive.ClientHttpConnector; import org.springframework.http.client.reactive.ReactorClientHttpConnector; import org.springframework.test.web.reactive.server.WebTestClient; import org.springframework.web.reactive.function.client.WebClient; +import org.springframework.web.server.ServerWebExchange; import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.GATEWAY_HANDLER_MAPPER_ATTR; import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.GATEWAY_ROUTE_ATTR; @@ -92,15 +94,8 @@ public class BaseWebClientTests { } @Bean - public GlobalFilter recursiveHttpbinFilter() { - return (exchange, chain) -> { - if (exchange.getRequest().getPath().toString() - .contains("httpbin/httpbin")) { - return Mono.error( - new IllegalStateException("recursive call to /httpbin")); - } - return chain.filter(exchange); - }; + public RecursiveHttpbinFilter recursiveHttpbinFilter() { + return new RecursiveHttpbinFilter(); } @Bean @@ -126,6 +121,19 @@ public class BaseWebClientTests { } + public static class RecursiveHttpbinFilter implements GlobalFilter { + + @Override + public Mono filter(ServerWebExchange exchange, GatewayFilterChain chain) { + if (exchange.getRequest().getPath().toString().contains("httpbin/httpbin")) { + return Mono + .error(new IllegalStateException("recursive call to /httpbin")); + } + return chain.filter(exchange); + } + + } + @EnableAutoConfiguration @SpringBootConfiguration @Import(DefaultTestConfig.class)