Move Kerberos proxy auth off the Netty event loop

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<Builder, BootstrapProvider> createBootStrapProvider) {
         this.configuration = builder.configuration;
@@ -105,6 +111,10 @@ private AwaitCloseChannelPoolMap(Builder builder, Function<Builder, BootstrapPro
         this.sslContextProvider = new SslContextProvider(configuration, protocol, protocolNegotiation, sslProvider);
         this.useNonBlockingDnsResolver = builder.useNonBlockingDnsResolver;
         this.negotiateAuthConfig = builder.negotiateAuthConfig;
+        // Both are resolved once per client rather than per pool: the proxy configuration is fixed for the life of the client,
+        // so there is no reason to duplicate the generator, or its executor, for every remote host.
+        this.proxyAuthExecutor = needsProxyAuthExecutor(proxyConfiguration) ? createProxyAuthExecutor() : null;
+        this.proxyAuthGenerator = resolveProxyAuthGenerator(proxyConfiguration, negotiateAuthConfig, proxyAuthExecutor);
     }
 
     private AwaitCloseChannelPoolMap(Builder builder) {
@@ -149,7 +159,7 @@ protected SimpleChannelPoolAwareChannelPool newPool(URI key) {
         if (shouldUseProxyForHost(key)) {
             tcpChannelPool = new BetterSimpleChannelPool(bootstrap, NOOP_HANDLER);
             baseChannelPool = new Http1TunnelConnectionPool(bootstrap.config().group().next(), tcpChannelPool, sslContext,
-                                            proxyAddress(key), resolveProxyAuthGenerator(proxyConfiguration),
+                                            proxyAddress(key), proxyAuthGenerator,
                                             key, pipelineInitializer, configuration);
         } else {
             tcpChannelPool = new BetterSimpleChannelPool(bootstrap, pipelineInitializer);
@@ -162,13 +172,34 @@ protected SimpleChannelPoolAwareChannelPool newPool(URI key) {
         return new SimpleChannelPoolAwareChannelPool(wrappedPool, tcpChannelPool);
     }
 
-    private ProxyAuthGenerator resolveProxyAuthGenerator(ProxyConfiguration proxyConfiguration) {
+    private static boolean needsProxyAuthExecutor(ProxyConfiguration proxyConfiguration) {
+        return proxyConfiguration != null && proxyConfiguration.proxyAuthScheme() == ProxyAuthScheme.NEGOTIATE;
+    }
+
+    /**
+     * Executor for auth schemes whose params are expensive to generate. A single thread, so generation is serialized: the
+     * point is only to keep the blocking work off the event loops, not to parallelize it. Daemon threaded so it never keeps
+     * the JVM alive, and shut down when this pool map is closed.
+     */
+    private static ExecutorService createProxyAuthExecutor() {
+        return Executors.newSingleThreadExecutor(new ThreadFactoryBuilder().threadNamePrefix("sdk-netty-proxy-auth")
+                                                                          .daemonThreads(true)
+                                                                          .build());
+    }
+
+    private static ProxyAuthGenerator resolveProxyAuthGenerator(ProxyConfiguration proxyConfiguration,
+                                                                Configuration negotiateAuthConfig,
+                                                                Executor proxyAuthExecutor) {
+        if (proxyConfiguration == null) {
+            return null;
+        }
+
         ProxyAuthScheme proxyAuthScheme = proxyConfiguration.proxyAuthScheme();
 
         if (proxyAuthScheme != null) {
             switch (proxyAuthScheme) {
                 case NEGOTIATE:
-                    return new NegotiateProxyAuthGenerator(negotiateAuthConfig);
+                    return new NegotiateProxyAuthGenerator(negotiateAuthConfig, proxyAuthExecutor);
                 case BASIC: {
                     String username = proxyConfiguration.username();
                     String password = proxyConfiguration.password();
@@ -204,6 +235,10 @@ public void close() {
         Collection<SimpleChannelPoolAwareChannelPool> channelPools = pools().values();
         super.close();
 
+        if (proxyAuthExecutor != null) {
+            proxyAuthExecutor.shutdownNow();
+        }
+
         try {
             CompletableFuture.allOf(channelPools.stream()
                                                 .map(pool -> pool.underlyingSimpleChannelPool().closeFuture())

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<String> 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)));
     }
 }

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<String> 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();
 

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.
+     * <p>
+     * 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<String> generateAuthParams(URI proxyEndpoint);
 }

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<String> 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;

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<Arguments> invalidCtorParams() {

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);
         };
 

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<Thread> 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");
+    }
+
 }