diff --git a/http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/AwaitCloseChannelPoolMap.java b/http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/AwaitCloseChannelPoolMap.java index d9441a2f6ee2..d2f74cd8dfa9 100644 --- a/http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/AwaitCloseChannelPoolMap.java +++ b/http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/AwaitCloseChannelPoolMap.java @@ -32,6 +32,9 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicReference; @@ -47,6 +50,7 @@ import software.amazon.awssdk.http.nio.netty.internal.http2.HttpOrHttp2ChannelPool; import software.amazon.awssdk.http.nio.netty.internal.utils.NettyClientLogger; import software.amazon.awssdk.utils.StringUtils; +import software.amazon.awssdk.utils.ThreadFactoryBuilder; /** * Implementation of {@link SdkChannelPoolMap} that awaits channel pools to be closed upon closing. @@ -91,6 +95,8 @@ public void channelCreated(Channel ch) throws Exception { private final Boolean useNonBlockingDnsResolver; private final Configuration negotiateAuthConfig; + private final ProxyAuthGenerator proxyAuthGenerator; + private final ExecutorService proxyAuthExecutor; private AwaitCloseChannelPoolMap(Builder builder, Function createBootStrapProvider) { this.configuration = builder.configuration; @@ -105,6 +111,10 @@ private AwaitCloseChannelPoolMap(Builder builder, Function channelPools = pools().values(); super.close(); + if (proxyAuthExecutor != null) { + proxyAuthExecutor.shutdownNow(); + } + try { CompletableFuture.allOf(channelPools.stream() .map(pool -> pool.underlyingSimpleChannelPool().closeFuture()) diff --git a/http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/BasicProxyAuthGenerator.java b/http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/BasicProxyAuthGenerator.java index 36055cf0b0fe..dfd22fc9437e 100644 --- a/http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/BasicProxyAuthGenerator.java +++ b/http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/BasicProxyAuthGenerator.java @@ -18,6 +18,7 @@ import io.netty.util.CharsetUtil; import java.net.URI; import java.util.Base64; +import java.util.concurrent.CompletableFuture; import software.amazon.awssdk.annotations.SdkInternalApi; import software.amazon.awssdk.http.nio.netty.ProxyAuthScheme; import software.amazon.awssdk.utils.Validate; @@ -43,8 +44,9 @@ public ProxyAuthScheme scheme() { } @Override - public String generateAuthParams(URI proxyEndpoint) { + public CompletableFuture generateAuthParams(URI proxyEndpoint) { + // Purely local and cheap, so this completes inline rather than hopping to another thread. String authToken = String.format("%s:%s", this.username, this.password); - return Base64.getEncoder().encodeToString(authToken.getBytes(CharsetUtil.UTF_8)); + return CompletableFuture.completedFuture(Base64.getEncoder().encodeToString(authToken.getBytes(CharsetUtil.UTF_8))); } } diff --git a/http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/NegotiateProxyAuthGenerator.java b/http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/NegotiateProxyAuthGenerator.java index 0866073946c2..67801981df85 100644 --- a/http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/NegotiateProxyAuthGenerator.java +++ b/http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/NegotiateProxyAuthGenerator.java @@ -20,6 +20,8 @@ import java.security.PrivilegedExceptionAction; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executor; import javax.security.auth.Subject; import javax.security.auth.login.AppConfigurationEntry; import javax.security.auth.login.Configuration; @@ -33,6 +35,7 @@ import software.amazon.awssdk.annotations.SdkInternalApi; import software.amazon.awssdk.http.nio.netty.ProxyAuthScheme; import software.amazon.awssdk.utils.BinaryUtils; +import software.amazon.awssdk.utils.Validate; /** * Auth generator for Kerberos. This does not login/authentication to Kerberos. It expects the ticket cache to be present and @@ -45,17 +48,20 @@ public class NegotiateProxyAuthGenerator implements ProxyAuthGenerator { private static final String OID = "1.3.6.1.5.5.2"; private static final String SERVICE_NAME = "HTTP"; private final Configuration config; + private final Executor executor; - public NegotiateProxyAuthGenerator() { - this(createDefaultConfig()); - } - - public NegotiateProxyAuthGenerator(Configuration config) { + /** + * @param config The JAAS configuration to log in with, or null to use the default ticket-cache-only configuration. + * @param executor Executor to run the blocking Kerberos work on. Must not be a Netty event loop; see + * {@link #generateAuthParams(URI)}. Its lifecycle is owned by the caller. + */ + public NegotiateProxyAuthGenerator(Configuration config, Executor executor) { if (config != null) { this.config = config; } else { this.config = createDefaultConfig(); } + this.executor = Validate.paramNotNull(executor, "executor"); } @Override @@ -64,7 +70,14 @@ public ProxyAuthScheme scheme() { } @Override - public String generateAuthParams(URI proxyEndpoint) { + public CompletableFuture generateAuthParams(URI proxyEndpoint) { + // Must not run on the caller's thread: the caller is a Netty event loop thread, and the work below reads the ticket + // cache from disk and may make a blocking TGS request to the KDC. Blocking the loop would stall every other channel + // assigned to it. + return CompletableFuture.supplyAsync(() -> generateAuthParamsBlocking(proxyEndpoint), executor); + } + + private String generateAuthParamsBlocking(URI proxyEndpoint) { try { Subject subject = getSubject(); diff --git a/http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/ProxyAuthGenerator.java b/http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/ProxyAuthGenerator.java index eeb84fbdb6f5..7924635f0620 100644 --- a/http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/ProxyAuthGenerator.java +++ b/http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/ProxyAuthGenerator.java @@ -16,6 +16,7 @@ package software.amazon.awssdk.http.nio.netty.internal; import java.net.URI; +import java.util.concurrent.CompletableFuture; import software.amazon.awssdk.annotations.SdkInternalApi; import software.amazon.awssdk.http.nio.netty.ProxyAuthScheme; @@ -31,6 +32,10 @@ public interface ProxyAuthGenerator { /** * Generate the auth params for this request. + *

+ * This is asynchronous because generating the params may block - Kerberos, for example, may need to read the ticket cache + * from disk and contact the KDC. Implementations that block MUST complete the returned future from a thread other than the + * caller's; the caller is a Netty event loop thread, and blocking it would stall every other channel assigned to that loop. */ - String generateAuthParams(URI proxyEndpoint); + CompletableFuture generateAuthParams(URI proxyEndpoint); } diff --git a/http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/ProxyTunnelInitHandler.java b/http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/ProxyTunnelInitHandler.java index aeda02f02e0f..db1be11492e3 100644 --- a/http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/ProxyTunnelInitHandler.java +++ b/http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/ProxyTunnelInitHandler.java @@ -31,6 +31,8 @@ import io.netty.util.concurrent.Promise; import java.io.IOException; import java.net.URI; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; import java.util.function.Supplier; import software.amazon.awssdk.annotations.SdkInternalApi; import software.amazon.awssdk.annotations.SdkTestInternalApi; @@ -95,9 +97,50 @@ public void handlerAdded(ChannelHandlerContext ctx) { ChannelPipeline pipeline = ctx.pipeline(); pipeline.addBefore(ctx.name(), null, httpCodecSupplier.get()); + if (authGenerator == null) { + sendConnectRequest(ctx, null); + return; + } + + CompletableFuture authParams; + try { + authParams = authGenerator.generateAuthParams(proxyAddress); + } catch (Throwable t) { + handleConnectRequestFailure(ctx, t); + return; + } + + // Basic auth completes inline, so only pay for a thread hop when the generator is actually asynchronous. When it is, + // the future completes on another thread, so hop back to the event loop before touching the channel or this + // handler's state. + boolean completesInline = authParams.isDone(); + authParams.whenComplete((params, error) -> { + if (completesInline) { + sendConnectRequestWithAuth(ctx, params, error); + } else { + ctx.executor().execute(() -> sendConnectRequestWithAuth(ctx, params, error)); + } + }); + } + + private void sendConnectRequestWithAuth(ChannelHandlerContext ctx, String authParams, Throwable error) { + // The channel may have gone away while we were waiting; whoever completed the promise has already cleaned up. + if (initPromise.isDone()) { + return; + } + + if (error != null) { + handleConnectRequestFailure(ctx, unwrap(error)); + return; + } + + sendConnectRequest(ctx, String.format("%s %s", authGenerator.scheme().value(), authParams)); + } + + private void sendConnectRequest(ChannelHandlerContext ctx, String proxyAuthorization) { HttpRequest connectRequest; try { - connectRequest = connectRequest(); + connectRequest = connectRequest(proxyAuthorization); } catch (Throwable t) { handleConnectRequestFailure(ctx, t); return; @@ -110,6 +153,10 @@ public void handlerAdded(ChannelHandlerContext ctx) { }); } + private static Throwable unwrap(Throwable t) { + return t instanceof CompletionException && t.getCause() != null ? t.getCause() : t; + } + @Override public void handlerRemoved(ChannelHandlerContext ctx) { if (ctx.pipeline().get(HttpClientCodec.class) != null) { @@ -170,15 +217,14 @@ private void closeAndRelease(ChannelHandlerContext ctx) { sourcePool.release(ctx.channel()); } - private HttpRequest connectRequest() { + private HttpRequest connectRequest(String proxyAuthorization) { String uri = getUri(); HttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.CONNECT, uri, Unpooled.EMPTY_BUFFER); request.headers().add(HttpHeaderNames.HOST, uri); - if (authGenerator != null) { - String auth = String.format("%s %s", authGenerator.scheme().value(), authGenerator.generateAuthParams(proxyAddress)); - request.headers().add(HttpHeaderNames.PROXY_AUTHORIZATION, auth); + if (proxyAuthorization != null) { + request.headers().add(HttpHeaderNames.PROXY_AUTHORIZATION, proxyAuthorization); } return request; diff --git a/http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/internal/BasicProxyAuthGeneratorTest.java b/http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/internal/BasicProxyAuthGeneratorTest.java index b0294ea768c3..b3f904f7e483 100644 --- a/http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/internal/BasicProxyAuthGeneratorTest.java +++ b/http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/internal/BasicProxyAuthGeneratorTest.java @@ -52,7 +52,7 @@ void generateAuthParams_generatedCorrectly() { .encodeToString(String.format("%s:%s", USERNAME, PASSWORD) .getBytes(StandardCharsets.UTF_8)); - assertThat(authGenerator.generateAuthParams(URI.create("http://amazon.com"))).isEqualTo(expected); + assertThat(authGenerator.generateAuthParams(URI.create("http://amazon.com")).join()).isEqualTo(expected); } private static Stream invalidCtorParams() { diff --git a/http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/internal/Http1TunnelConnectionPoolTest.java b/http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/internal/Http1TunnelConnectionPoolTest.java index b61ac05e0b10..b06eee6eaadc 100644 --- a/http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/internal/Http1TunnelConnectionPoolTest.java +++ b/http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/internal/Http1TunnelConnectionPoolTest.java @@ -264,7 +264,7 @@ public void proxyAuthProvided_addInitHandler_withAuth(){ Http1TunnelConnectionPool.InitHandlerSupplier supplier = (srcPool, proxyEndpoint, proxyAuthGenerator, remoteAddr, initFuture) -> { initFuture.setSuccess(mockChannel); - data.authHeader = proxyAuthGenerator.generateAuthParams(proxyEndpoint); + data.authHeader = proxyAuthGenerator.generateAuthParams(proxyEndpoint).join(); return mock(ChannelHandler.class); }; diff --git a/http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/internal/NegotiateProxyAuthGeneratorTest.java b/http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/internal/NegotiateProxyAuthGeneratorTest.java index 218630a46ac4..97f2c3c2f4e1 100644 --- a/http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/internal/NegotiateProxyAuthGeneratorTest.java +++ b/http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/internal/NegotiateProxyAuthGeneratorTest.java @@ -26,6 +26,10 @@ import java.nio.file.Path; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.CompletionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicReference; import javax.security.auth.login.AppConfigurationEntry; import javax.security.auth.login.Configuration; import org.apache.kerby.kerberos.kerb.KrbException; @@ -39,6 +43,9 @@ public class NegotiateProxyAuthGeneratorTest { private static final String KRB5_PROP = "java.security.krb5.conf"; + private static final String EXECUTOR_THREAD_NAME = "test-proxy-auth"; + private static final ExecutorService executor = + Executors.newSingleThreadExecutor(r -> new Thread(r, EXECUTOR_THREAD_NAME)); private static Path tempDir; private static Path keytabFile; private static Path ccacheFile; @@ -104,6 +111,7 @@ public AppConfigurationEntry[] getAppConfigurationEntry(String name) { @AfterAll static void teardown() throws KrbException { + executor.shutdownNow(); if (krb5PropSave != null) { System.setProperty(KRB5_PROP, krb5PropSave); } else { @@ -115,11 +123,11 @@ static void teardown() throws KrbException { @Test void generateAuthParams_configValid_successfullyGeneratesToken() { - NegotiateProxyAuthGenerator authGenerator = new NegotiateProxyAuthGenerator(config); + NegotiateProxyAuthGenerator authGenerator = new NegotiateProxyAuthGenerator(config, executor); URI proxyEndpoint = URI.create("https://localhost:8192"); - assertThat(authGenerator.generateAuthParams(proxyEndpoint)).startsWith("YII"); + assertThat(authGenerator.generateAuthParams(proxyEndpoint).join()).startsWith("YII"); } @Test @@ -140,12 +148,39 @@ public AppConfigurationEntry[] getAppConfigurationEntry(String name) { } }; - NegotiateProxyAuthGenerator authGenerator = new NegotiateProxyAuthGenerator(missingCacheConfig); + NegotiateProxyAuthGenerator authGenerator = new NegotiateProxyAuthGenerator(missingCacheConfig, executor); - assertThatThrownBy(() -> authGenerator.generateAuthParams(URI.create("https://localhost:8192"))) - .isInstanceOf(RuntimeException.class) + assertThatThrownBy(() -> authGenerator.generateAuthParams(URI.create("https://localhost:8192")).join()) + .isInstanceOf(CompletionException.class) + .hasCauseInstanceOf(RuntimeException.class) .hasMessageContaining("kinit") .hasMessageContaining("ticket cache"); } + @Test + void generateAuthParams_runsOnSuppliedExecutor() { + AtomicReference loginThread = new AtomicReference<>(); + Configuration recordingConfig = new Configuration() { + @Override + public AppConfigurationEntry[] getAppConfigurationEntry(String name) { + loginThread.set(Thread.currentThread()); + return config.getAppConfigurationEntry(name); + } + }; + + new NegotiateProxyAuthGenerator(recordingConfig, executor).generateAuthParams(URI.create("https://localhost:8192")) + .join(); + + // The blocking Kerberos work must never run on the caller's thread, which in production is a Netty event loop. + assertThat(loginThread.get()).isNotSameAs(Thread.currentThread()); + assertThat(loginThread.get().getName()).isEqualTo(EXECUTOR_THREAD_NAME); + } + + @Test + void constructor_nullExecutor_throws() { + assertThatThrownBy(() -> new NegotiateProxyAuthGenerator(config, null)) + .isInstanceOf(NullPointerException.class) + .hasMessageContaining("executor"); + } + } diff --git a/http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/internal/ProxyTunnelInitHandlerTest.java b/http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/internal/ProxyTunnelInitHandlerTest.java index 143cc174701f..c0627ccbdef2 100644 --- a/http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/internal/ProxyTunnelInitHandlerTest.java +++ b/http-clients/netty-nio-client/src/test/java/software/amazon/awssdk/http/nio/netty/internal/ProxyTunnelInitHandlerTest.java @@ -42,10 +42,12 @@ import io.netty.handler.ssl.SslCloseCompletionEvent; import io.netty.handler.ssl.SslHandler; import io.netty.util.CharsetUtil; +import io.netty.util.concurrent.EventExecutor; import io.netty.util.concurrent.Promise; import java.io.IOException; import java.net.URI; import java.util.Base64; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.function.Supplier; import org.junit.AfterClass; @@ -245,7 +247,6 @@ public void handlerAdded_writesRequest_withAuth() { @Test public void handlerAdded_authParamsGeneratorThrows_failsFuture() { ProxyAuthGenerator authGenerator = mock(ProxyAuthGenerator.class); - when(authGenerator.scheme()).thenReturn(ProxyAuthScheme.BASIC); when(authGenerator.generateAuthParams(any(URI.class))).thenThrow(new RuntimeException("auth generator error")); Promise promise = GROUP.next().newPromise(); @@ -259,6 +260,55 @@ public void handlerAdded_authParamsGeneratorThrows_failsFuture() { .hasRootCauseMessage("auth generator error"); } + @Test + public void handlerAdded_authParamsCompleteAsynchronously_writesRequestOnceParamsAvailable() throws Exception { + CompletableFuture authParams = new CompletableFuture<>(); + ProxyAuthGenerator authGenerator = mock(ProxyAuthGenerator.class); + when(authGenerator.scheme()).thenReturn(ProxyAuthScheme.NEGOTIATE); + when(authGenerator.generateAuthParams(any(URI.class))).thenReturn(authParams); + + EventExecutor executor = GROUP.next(); + when(mockCtx.executor()).thenReturn(executor); + + Promise promise = GROUP.next().newPromise(); + ProxyTunnelInitHandler handler = new ProxyTunnelInitHandler(mockChannelPool, URI.create("https://proxy.com"), + authGenerator, REMOTE_HOST, promise); + handler.handlerAdded(mockCtx); + + // The calling thread must not have waited for the auth params, so nothing is written yet. + verify(mockChannel, never()).writeAndFlush(any()); + + authParams.complete("token"); + // Drain the executor so the queued continuation has run. + executor.submit(() -> { }).get(); + + ArgumentCaptor requestCaptor = ArgumentCaptor.forClass(HttpRequest.class); + verify(mockChannel).writeAndFlush(requestCaptor.capture()); + assertThat(requestCaptor.getValue().headers().get(HttpHeaderNames.PROXY_AUTHORIZATION)).isEqualTo("Negotiate token"); + } + + @Test + public void handlerAdded_authParamsFailAsynchronously_failsFuture() throws Exception { + CompletableFuture authParams = new CompletableFuture<>(); + ProxyAuthGenerator authGenerator = mock(ProxyAuthGenerator.class); + when(authGenerator.generateAuthParams(any(URI.class))).thenReturn(authParams); + + EventExecutor executor = GROUP.next(); + when(mockCtx.executor()).thenReturn(executor); + + Promise promise = GROUP.next().newPromise(); + ProxyTunnelInitHandler handler = new ProxyTunnelInitHandler(mockChannelPool, URI.create("https://proxy.com"), + authGenerator, REMOTE_HOST, promise); + handler.handlerAdded(mockCtx); + + authParams.completeExceptionally(new RuntimeException("auth generator error")); + executor.submit(() -> { }).get(); + + verify(mockChannel, never()).writeAndFlush(any()); + assertThatThrownBy(promise::get).hasMessageContaining("Unable to send CONNECT request to proxy") + .hasRootCauseMessage("auth generator error"); + } + private void successResponse(ProxyTunnelInitHandler handler) { DefaultHttpResponse resp = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK); handler.channelRead(mockCtx, resp);