Implement HTTP proxy CONNECT with ALPN

servicetalk-grpc-netty/src/test/java/io/servicetalk/grpc/netty/GrpcProxyTunnelTest.java

@@ -0,0 +1,148 @@
+/*
+ * Copyright © 2023 Apple Inc. and the ServiceTalk project 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
+ *
+ *   http://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 io.servicetalk.grpc.netty;
+
+import io.servicetalk.concurrent.api.Single;
+import io.servicetalk.context.api.ContextMap.Key;
+import io.servicetalk.grpc.api.DefaultGrpcClientMetadata;
+import io.servicetalk.grpc.api.GrpcClientMetadata;
+import io.servicetalk.grpc.api.GrpcStatusException;
+import io.servicetalk.http.api.StreamingHttpConnectionFilter;
+import io.servicetalk.http.api.StreamingHttpRequest;
+import io.servicetalk.http.api.StreamingHttpResponse;
+import io.servicetalk.http.netty.ProxyResponseException;
+import io.servicetalk.http.netty.ProxyTunnel;
+import io.servicetalk.test.resources.DefaultTestCerts;
+import io.servicetalk.transport.api.ClientSslConfigBuilder;
+import io.servicetalk.transport.api.ConnectionInfo;
+import io.servicetalk.transport.api.HostAndPort;
+import io.servicetalk.transport.api.ServerContext;
+import io.servicetalk.transport.api.ServerSslConfigBuilder;
+
+import io.grpc.examples.helloworld.Greeter;
+import io.grpc.examples.helloworld.Greeter.BlockingGreeterClient;
+import io.grpc.examples.helloworld.HelloReply;
+import io.grpc.examples.helloworld.HelloRequest;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.InetSocketAddress;
+
+import static io.servicetalk.context.api.ContextMap.Key.newKey;
+import static io.servicetalk.grpc.api.GrpcStatusCode.UNKNOWN;
+import static io.servicetalk.http.api.HttpProtocolVersion.HTTP_2_0;
+import static io.servicetalk.http.api.HttpResponseStatus.INTERNAL_SERVER_ERROR;
+import static io.servicetalk.http.api.HttpResponseStatus.PROXY_AUTHENTICATION_REQUIRED;
+import static io.servicetalk.test.resources.DefaultTestCerts.serverPemHostname;
+import static io.servicetalk.transport.netty.internal.AddressUtils.localAddress;
+import static io.servicetalk.transport.netty.internal.AddressUtils.serverHostAndPort;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class GrpcProxyTunnelTest {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(GrpcProxyTunnelTest.class);
+    private static final String AUTH_TOKEN = "aGVsbG86d29ybGQ=";
+    private static final String GREETING_PREFIX = "Hello ";
+    private static final Key<ConnectionInfo> CONNECTION_INFO_KEY =
+            newKey("CONNECTION_INFO_KEY", ConnectionInfo.class);
+
+    private final ProxyTunnel proxyTunnel;
+    private final HostAndPort proxyAddress;
+    private final ServerContext serverContext;
+    private final BlockingGreeterClient client;
+
+    GrpcProxyTunnelTest() throws Exception {
+        proxyTunnel = new ProxyTunnel();
+        proxyAddress = proxyTunnel.startProxy();
+        serverContext = GrpcServers.forAddress(localAddress(0))
+                .initializeHttp(httpBuilder -> httpBuilder
+                        .sslConfig(new ServerSslConfigBuilder(DefaultTestCerts::loadServerPem,
+                                DefaultTestCerts::loadServerKey).build()))
+                .listenAndAwait((Greeter.BlockingGreeterService) (ctx, request) ->
+                        HelloReply.newBuilder().setMessage(GREETING_PREFIX + request.getName()).build());
+        client = GrpcClients.forAddress(serverHostAndPort(serverContext))
+                .initializeHttp(httpBuilder -> httpBuilder.proxyAddress(proxyAddress)
+                        .sslConfig(new ClientSslConfigBuilder(DefaultTestCerts::loadServerCAPem)
+                                .peerHost(serverPemHostname()).build())
+                        .appendConnectionFilter(connection -> new StreamingHttpConnectionFilter(connection) {
+                            @Override
+                            public Single<StreamingHttpResponse> request(final StreamingHttpRequest request) {
+                                return delegate().request(request)
+                                        .whenOnSuccess(response -> response.context()
+                                                .put(CONNECTION_INFO_KEY, connection.connectionContext()));
+                            }
+                        }))
+                .buildBlocking(new Greeter.ClientFactory());
+    }
+
+    @AfterEach
+    void tearDown() throws Exception {
+        safeClose(client);
+        safeClose(serverContext);
+        safeClose(proxyTunnel);
+    }
+
+    private static void safeClose(AutoCloseable closeable) {
+        try {
+            closeable.close();
+        } catch (Exception e) {
+            LOGGER.error("Unexpected exception while closing", e);
+        }
+    }
+
+    @Test
+    void testRequest() throws Exception {
+        String name = "foo";
+        GrpcClientMetadata metaData = new DefaultGrpcClientMetadata();
+        HelloReply reply = client.sayHello(metaData, HelloRequest.newBuilder().setName(name).build());
+        assertThat(reply.getMessage(), is(GREETING_PREFIX + name));
+        ConnectionInfo connectionInfo = metaData.responseContext().get(CONNECTION_INFO_KEY);
+        assertThat(connectionInfo, is(notNullValue()));
+        assertThat(connectionInfo.protocol(), is(HTTP_2_0));
+        assertThat(connectionInfo.sslConfig(), is(notNullValue()));
+        assertThat(connectionInfo.sslSession(), is(notNullValue()));
+        assertThat(((InetSocketAddress) connectionInfo.remoteAddress()).getPort(), is(proxyAddress.port()));
+        assertThat(proxyTunnel.connectCount(), is(1));
+    }
+
+    @Test
+    void testProxyAuthRequired() throws Exception {
+        proxyTunnel.basicAuthToken(AUTH_TOKEN);
+        GrpcStatusException e = assertThrows(GrpcStatusException.class,
+                () -> client.sayHello(HelloRequest.newBuilder().setName("foo").build()));
+        assertThat(e.status().code(), is(UNKNOWN));
+        Throwable cause = e.getCause();
+        assertThat(cause, is(instanceOf(ProxyResponseException.class)));
+        assertThat(((ProxyResponseException) cause).status(), is(PROXY_AUTHENTICATION_REQUIRED));
+    }
+
+    @Test
+    void testBadProxyResponse() throws Exception {
+        proxyTunnel.badResponseProxy();
+        GrpcStatusException e = assertThrows(GrpcStatusException.class,
+                () -> client.sayHello(HelloRequest.newBuilder().setName("foo").build()));
+        assertThat(e.status().code(), is(UNKNOWN));
+        Throwable cause = e.getCause();
+        assertThat(cause, is(instanceOf(ProxyResponseException.class)));
+        assertThat(((ProxyResponseException) cause).status(), is(INTERNAL_SERVER_ERROR));
+    }
+}

servicetalk-http-api/src/main/java/io/servicetalk/http/api/SingleAddressHttpClientBuilder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright © 2018-2019 Apple Inc. and the ServiceTalk project authors
+ * Copyright © 2018-2023 Apple Inc. and the ServiceTalk project authors
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -53,6 +53,12 @@ public interface SingleAddressHttpClientBuilder<U, R> extends HttpClientBuilder<
      * If the client talks to a proxy over http (not https, {@link #sslConfig(ClientSslConfig) ClientSslConfig} is NOT
      * configured), it will rewrite the request-target to
      * <a href="https://tools.ietf.org/html/rfc7230#section-5.3.2">absolute-form</a>, as specified by the RFC.
+     * <p>
+     * For secure proxy tunnels (when {@link #sslConfig(ClientSslConfig) ClientSslConfig} is configured) the tunnel is
+     * always initialized using
+     * <a href="https://datatracker.ietf.org/doc/html/rfc9110#section-9.3.6">HTTP/1.1 CONNECT</a> request. The actual
+     * protocol will be negotiated via <a href="https://tools.ietf.org/html/rfc7301">ALPN extension</a> of TLS protocol,
+     * taking into account HTTP protocols configured via {@link #protocols(HttpProtocolConfig...)} method.
      *
      * @param proxyAddress Unresolved address of the proxy. When used with a builder created for a resolved address,
      * {@code proxyAddress} should also be already resolved – otherwise runtime exceptions may occur.

servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/AlpnChannelSingle.java

@@ -31,28 +31,30 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.function.Consumer;
 import javax.annotation.Nullable;
 import javax.net.ssl.SSLException;
 
 import static io.servicetalk.http.netty.AlpnIds.HTTP_1_1;
 import static io.servicetalk.transport.netty.internal.ChannelCloseUtils.assignConnectionError;
+import static java.util.Objects.requireNonNull;
 
 /**
  * A {@link Single} that initializes ALPN handler and completes after protocol negotiation.
  */
 final class AlpnChannelSingle extends ChannelInitSingle<String> {
-    private final boolean forceChannelRead;
+    private final Consumer<ChannelHandlerContext> onHandlerAdded;
 
     AlpnChannelSingle(final Channel channel,
                       final ChannelInitializer channelInitializer,
-                      final boolean forceChannelRead) {
+                      final Consumer<ChannelHandlerContext> onHandlerAdded) {
         super(channel, channelInitializer);
-        this.forceChannelRead = forceChannelRead;
+        this.onHandlerAdded = requireNonNull(onHandlerAdded);
     }
 
     @Override
     protected ChannelHandler newChannelHandler(final Subscriber<? super String> subscriber) {
-        return new AlpnChannelHandler(subscriber, forceChannelRead);
+        return new AlpnChannelHandler(subscriber, onHandlerAdded);
     }
 
     /**
@@ -65,24 +67,19 @@ private static final class AlpnChannelHandler extends ApplicationProtocolNegotia
 
         @Nullable
         private SingleSource.Subscriber<? super String> subscriber;
-        private final boolean forceRead;
+        private final Consumer<ChannelHandlerContext> onHandlerAdded;
 
         AlpnChannelHandler(final SingleSource.Subscriber<? super String> subscriber,
-                           final boolean forceRead) {
+                           final Consumer<ChannelHandlerContext> onHandlerAdded) {
             super(HTTP_1_1);
             this.subscriber = subscriber;
-            this.forceRead = forceRead;
+            this.onHandlerAdded = onHandlerAdded;
         }
 
         @Override
         public void handlerAdded(final ChannelHandlerContext ctx) throws Exception {
             super.handlerAdded(ctx);
-            if (forceRead) {
-                // Force a read to get the SSL handshake started. We initialize pipeline before
-                // SslHandshakeCompletionEvent will complete, therefore, no data will be propagated before we finish
-                // initialization.
-                ctx.read();
-            }
+            onHandlerAdded.accept(ctx);
         }
 
         @Override

servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/AlpnLBHttpConnectionFactory.java

@@ -69,8 +69,8 @@ Single<FilterableStreamingHttpConnection> newFilterableConnection(
     private Single<FilterableStreamingHttpConnection> createConnection(
             final Channel channel, final ConnectionObserver connectionObserver,
             final ReadOnlyTcpClientConfig tcpConfig) {
-        return new AlpnChannelSingle(channel,
-                new TcpClientChannelInitializer(tcpConfig, connectionObserver), false).flatMap(protocol -> {
+        return new AlpnChannelSingle(channel, new TcpClientChannelInitializer(tcpConfig, connectionObserver),
+                ctx -> { /* SslHandler will automatically start handshake on channelActive */ }).flatMap(protocol -> {
             switch (protocol) {
                 case HTTP_1_1:
                     final H1ProtocolConfig h1Config = config.h1Config();
@@ -89,8 +89,12 @@ private Single<FilterableStreamingHttpConnection> createConnection(
                             new H2ClientParentChannelInitializer(h2Config),
                             connectionObserver, config.allowDropTrailersReadFromTransport());
                 default:
-                    return failed(new IllegalStateException("Unknown ALPN protocol negotiated: " + protocol));
+                    return unknownAlpnProtocol(protocol);
             }
         });
     }
+
+    static Single<FilterableStreamingHttpConnection> unknownAlpnProtocol(final String protocol) {
+        return failed(new IllegalStateException("Unknown ALPN protocol negotiated: " + protocol));
+    }
 }

servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/DefaultSingleAddressHttpClientBuilder.java

@@ -232,8 +232,9 @@ public HttpExecutionStrategy executionStrategy() {
                 return computedStrategy;
             }
         };
-        if (roConfig.h2Config() != null && roConfig.hasProxy()) {
-            throw new IllegalStateException("Proxying is not yet supported with HTTP/2");
+        final SslContext sslContext = roConfig.tcpConfig().sslContext();
+        if (roConfig.hasProxy() && sslContext == null && roConfig.h2Config() != null) {
+            throw new IllegalStateException("Proxying is not yet supported with plaintext HTTP/2");
         }
 
         // Track resources that potentially need to be closed when an exception is thrown during buildStreaming
@@ -247,7 +248,6 @@ public HttpExecutionStrategy executionStrategy() {
             final ExecutionStrategy connectionFactoryStrategy =
                     ctx.builder.strategyComputation.buildForConnectionFactory();
 
-            final SslContext sslContext = roConfig.tcpConfig().sslContext();
             if (roConfig.hasProxy() && sslContext != null) {
                 assert roConfig.connectAddress() != null;
                 final ConnectionFactoryFilter<R, FilterableStreamingHttpConnection> proxy =
@@ -266,14 +266,14 @@ public HttpExecutionStrategy executionStrategy() {
                     ctx.builder.addIdleTimeoutConnectionFilter ?
                             appendConnectionFilter(ctx.builder.connectionFilterFactory, DEFAULT_IDLE_TIMEOUT_FILTER) :
                             ctx.builder.connectionFilterFactory;
-            if (roConfig.isH2PriorKnowledge()) {
+            if (!roConfig.hasProxy() && roConfig.isH2PriorKnowledge()) {
                 H2ProtocolConfig h2Config = roConfig.h2Config();
                 assert h2Config != null;
                 connectionFactory = new H2LBHttpConnectionFactory<>(roConfig, executionContext,
                         connectionFilterFactory, reqRespFactory,
                         connectionFactoryStrategy, connectionFactoryFilter,
                         ctx.builder.loadBalancerFactory::toLoadBalancedConnection);
-            } else if (roConfig.tcpConfig().preferredAlpnProtocol() != null) {
+            } else if (!roConfig.hasProxy() && roConfig.tcpConfig().preferredAlpnProtocol() != null) {
                 H1ProtocolConfig h1Config = roConfig.h1Config();
                 H2ProtocolConfig h2Config = roConfig.h2Config();
                 connectionFactory = new AlpnLBHttpConnectionFactory<>(roConfig, executionContext,

servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/DeferredServerChannelBinder.java

@@ -31,6 +31,7 @@
 import io.servicetalk.transport.netty.internal.NettyConnectionContext;
 
 import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -90,8 +91,11 @@ private static Single<NettyConnectionContext> alpnInitChannel(final SocketAddres
                                                                   final StreamingHttpService service,
                                                                   final boolean drainRequestPayloadBody,
                                                                   final ConnectionObserver observer) {
-        return new AlpnChannelSingle(channel,
-                new TcpServerChannelInitializer(config.tcpConfig(), observer), true).flatMap(protocol -> {
+        return new AlpnChannelSingle(channel, new TcpServerChannelInitializer(config.tcpConfig(), observer),
+                // Force a read to get the SSL handshake started. We initialize pipeline before
+                // SslHandshakeCompletionEvent will complete, therefore, no data will be propagated before we finish
+                // initialization.
+                ChannelHandlerContext::read).flatMap(protocol -> {
             switch (protocol) {
                 case HTTP_1_1:
                     return NettyHttpServer.initChannel(channel, httpExecutionContext, config,

servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/HttpClientChannelInitializer.java

@@ -23,15 +23,22 @@
 
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelPipeline;
 
 import java.util.ArrayDeque;
+import java.util.List;
 import java.util.Queue;
 
 import static java.lang.Math.min;
+import static java.util.Arrays.asList;
+import static java.util.Collections.unmodifiableList;
 
 final class HttpClientChannelInitializer implements ChannelInitializer {
 
+    private static final List<Class<? extends ChannelHandler>> HANDLERS = unmodifiableList(asList(
+            HttpRequestEncoder.class, HttpResponseDecoder.class, CopyByteBufHandlerChannelInitializer.handlerClass()));
+
     private final ChannelInitializer delegate;
 
     /**
@@ -61,4 +68,15 @@ final class HttpClientChannelInitializer implements ChannelInitializer {
     public void init(final Channel channel) {
         delegate.init(channel);
     }
+
+    /**
+     * A list of {@link ChannelHandler} classes added to the {@link ChannelPipeline} in reverse order
+     * (from last to first).
+     *
+     * @return A list of {@link ChannelHandler} classes added to the {@link ChannelPipeline} in reverse order
+     * (from last to first).
+     */
+    static List<Class<? extends ChannelHandler>> handlers() {
+        return HANDLERS;
+    }
 }

servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/PipelinedLBHttpConnectionFactory.java

@@ -47,7 +47,8 @@ final class PipelinedLBHttpConnectionFactory<ResolvedAddress> extends AbstractLB
     Single<FilterableStreamingHttpConnection> newFilterableConnection(final ResolvedAddress resolvedAddress,
                                                                       final TransportObserver observer) {
         assert config.h1Config() != null;
-        return buildStreaming(executionContext, resolvedAddress, config, observer)
+        return buildStreaming(executionContext, resolvedAddress, config.tcpConfig(), config.h1Config(),
+                config.hasProxy(), observer)
                 .map(conn -> new PipelinedStreamingHttpConnection(conn, config.h1Config(),
                         reqRespFactoryFunc.apply(HTTP_1_1), config.allowDropTrailersReadFromTransport()));
     }