The HTTP client request path when resolving an address does not apply the default proxy options nor performs proxy filering.

Apply those after we resolve the address to a socket address before interracting with the HTTP connection manager.

Author: vietj

src/main/java/io/vertx/core/http/impl/HttpClientImpl.java

@@ -325,13 +325,13 @@ private Future<HttpClientRequest> doRequest(
     ContextInternal ctx = vertx.getOrCreateContext();
     ContextInternal connCtx = ctx.isEventLoopContext() ? ctx : vertx.createEventLoopContext(ctx.nettyEventLoop(), ctx.workerPool(), ctx.classLoader());
     Promise<HttpClientRequest> promise = ctx.promise();
-    Future<HttpClientStream> future;
-    ProxyOptions proxyOptions;
+    Future<ConnectionObtainedResult> future;
     if (endpointResolver != null && endpointResolver.accepts(server) != null) {
       Future<EndpointLookup> fut = endpointResolver.lookupEndpoint(ctx, server);
       future = fut.compose(lookup -> {
         SocketAddress address = lookup.address();
-        EndpointKey key = new EndpointKey(useSSL, sslOptions, proxyConfig, address, authority != null ? authority : HostAndPort.create(address.host(), address.port()));
+        ProxyOptions proxyOptions = computeProxyOptions(proxyConfig, address);
+        EndpointKey key = new EndpointKey(useSSL, sslOptions, proxyOptions, address, authority != null ? authority : HostAndPort.create(address.host(), address.port()));
         return httpCM.withEndpointAsync(key, httpEndpointProvider(), (endpoint, created) -> {
           Future<Lease<HttpClientConnection>> fut2 = endpoint.requestConnection(connCtx, connectTimeout);
           if (fut2 == null) {
@@ -347,19 +347,14 @@ private Future<HttpClientRequest> doRequest(
               return conn.createStream(ctx).map(stream -> {
                 HttpClientStream wrapped = new StatisticsGatheringHttpClientStream(stream, endpointRequest);
                 wrapped.closeHandler(v -> lease.recycle());
-                return wrapped;
+                return new ConnectionObtainedResult(proxyOptions, wrapped);
               });
             });
           }
         });
       });
-      if (future != null) {
-        proxyOptions = proxyConfig;
-      } else {
-        proxyOptions = null;
-      }
     } else if (server instanceof SocketAddress) {
-      proxyOptions = computeProxyOptions(proxyConfig, (SocketAddress) server);
+      ProxyOptions proxyOptions = computeProxyOptions(proxyConfig, (SocketAddress) server);
       EndpointKey key = new EndpointKey(useSSL, sslOptions, proxyOptions, (SocketAddress) server, authority);
       future = httpCM.withEndpointAsync(key, httpEndpointProvider(), (endpoint, created) -> {
         Future<Lease<HttpClientConnection>> fut = endpoint.requestConnection(connCtx, connectTimeout);
@@ -368,13 +363,11 @@ private Future<HttpClientRequest> doRequest(
         } else {
           return fut.compose(lease -> {
             HttpClientConnection conn = lease.get();
-            return conn.createStream(ctx).andThen(ar -> {
-              if (ar.succeeded()) {
-                HttpClientStream stream = ar.result();
-                stream.closeHandler(v -> {
-                  lease.recycle();
-                });
-              }
+            return conn.createStream(ctx).map(stream -> {
+              stream.closeHandler(v -> {
+                lease.recycle();
+              });
+              return new ConnectionObtainedResult(proxyOptions, stream);
             });
           });
         }
@@ -385,13 +378,22 @@ private Future<HttpClientRequest> doRequest(
     if (future == null) {
       return connCtx.failedFuture("Cannot resolve address " + server);
     } else {
-      future.map(stream -> {
-        return createRequest(stream, method, headers, requestURI, proxyOptions, useSSL, idleTimeout, followRedirects, traceOperation);
+      future.map(res -> {
+        return createRequest(res.stream, method, headers, requestURI, res.proxyOptions, useSSL, idleTimeout, followRedirects, traceOperation);
       }).onComplete(promise);
       return promise.future();
     }
   }
 
+  private static class ConnectionObtainedResult {
+    private final ProxyOptions proxyOptions;
+    private final HttpClientStream stream;
+    public ConnectionObtainedResult(ProxyOptions proxyOptions, HttpClientStream stream) {
+      this.proxyOptions = proxyOptions;
+      this.stream = stream;
+    }
+  }
+
   Future<HttpClientRequest> createRequest(HttpClientConnection conn, ContextInternal context) {
     return conn.createStream(context).map(this::createRequest);
   }

src/test/java/io/vertx/core/http/ResolvingHttpClientTest.java

@@ -1,7 +1,10 @@
 package io.vertx.core.http;
 
 import io.vertx.core.Future;
+import io.vertx.core.VertxOptions;
 import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.impl.CleanableHttpClient;
+import io.vertx.core.http.impl.HttpClientImpl;
 import io.vertx.core.http.impl.HttpClientInternal;
 import io.vertx.core.net.*;
 import io.vertx.test.core.VertxTestBase;
@@ -26,6 +29,17 @@ public class ResolvingHttpClientTest extends VertxTestBase {
   private BiConsumer<Integer, HttpServerRequest> requestHandler;
   private List<HttpServer> servers;
 
+  @Override
+  protected VertxOptions getOptions() {
+    VertxOptions options = super.getOptions();
+    options.getAddressResolverOptions().setHostsValue(Buffer.buffer("" +
+      "127.0.0.1 localhost\n" +
+      "127.0.0.1 s1.example.com\n" +
+      "127.0.0.1 s2.example.com\n"
+    ));
+    return options;
+  }
+
   private void startServers(int numServers) throws Exception {
     startServers(numServers, new HttpServerOptions());
   }
@@ -172,6 +186,59 @@ public void testResolveToSameSocketAddressWithProxy() throws Exception {
     assertNotNull(proxy.lastLocalAddress());
   }
 
+  @Test
+  public void testAcceptProxyFilter() throws Exception {
+    testFilter(true);
+  }
+
+  @Test
+  public void testRejectProxyFilter() throws Exception {
+    testFilter(false);
+  }
+
+  private void testFilter(boolean accept) throws Exception {
+    HttpProxy proxy = new HttpProxy();
+    proxy.start(vertx);
+    try {
+      int numServers = 2;
+      waitFor(numServers * 2);
+      startServers(numServers);
+      requestHandler = (idx, req) -> {
+        boolean proxied = idx == 0 && accept;
+        SocketAddress remote = req.connection().remoteAddress();
+        assertEquals(proxied, proxy.localAddresses().contains(remote.host() + ":" + remote.port()));
+        req.response().end("server-" + idx);
+      };
+      FakeAddressResolver resolver = new FakeAddressResolver();
+      resolver.registerAddress("example.com", Arrays.asList(
+        SocketAddress.inetSocketAddress(HttpTestBase.DEFAULT_HTTP_PORT, "s1.example.com"),
+        SocketAddress.inetSocketAddress(HttpTestBase.DEFAULT_HTTP_PORT + 1, "s2.example.com")));
+      HttpClientInternal client = (HttpClientInternal) vertx.httpClientBuilder()
+        .with(new HttpClientOptions()
+          .setProxyOptions(new ProxyOptions().setType(ProxyType.HTTP).setHost("localhost").setPort(proxy.port())))
+        .withAddressResolver(resolver)
+        .build();
+      ((HttpClientImpl)((CleanableHttpClient)client).delegate).proxyFilter(so -> {
+        return accept && so.host().equals("s1.example.com");
+      });
+      Set<String> responses = Collections.synchronizedSet(new HashSet<>());
+      for (int i = 0;i < numServers * 2;i++) {
+        client.request(new RequestOptions().setServer(new FakeAddress("example.com"))).compose(req -> req
+          .send()
+          .andThen(onSuccess(resp -> assertEquals(200, resp.statusCode())))
+          .compose(HttpClientResponse::body)
+        ).onComplete(onSuccess(v -> {
+          responses.add(v.toString());
+          complete();
+        }));
+      }
+      await();
+      Assert.assertEquals(new HashSet<>(Arrays.asList("server-0", "server-1")), responses);
+    } finally {
+      proxy.stop();
+    }
+  }
+
   @Test
   public void testResolveFailure() {
     Exception cause = new Exception("Not found");
@@ -539,5 +606,4 @@ private void testSSL(RequestOptions request, boolean expectFailure, boolean veri
       assertTrue(expectFailure);
     }
   }
-
 }