[fix][sec] Upgrade to async-http-client 2.14.5 to address CVE-2026-40490 (#25546)

lhotari · lhotari · commit cfa8192198c2 · 2026-04-17T16:26:39.000+03:00
(cherry picked from commit a1613bc)
diff --git a/distribution/server/src/assemble/LICENSE.bin.txt b/distribution/server/src/assemble/LICENSE.bin.txt
@@ -392,8 +392,8 @@ The Apache Software License, Version 2.0
  * AirCompressor
     - io.airlift-aircompressor-2.0.3.jar
  * AsyncHttpClient
-    - org.asynchttpclient-async-http-client-2.12.4.jar
-    - org.asynchttpclient-async-http-client-netty-utils-2.12.4.jar
+    - org.asynchttpclient-async-http-client-2.14.5.jar
+    - org.asynchttpclient-async-http-client-netty-utils-2.14.5.jar
  * Jetty
     - org.eclipse.jetty-jetty-client-9.4.58.v20250814.jar
     - org.eclipse.jetty-jetty-continuation-9.4.58.v20250814.jar
@@ -611,7 +611,7 @@ Eclipse Public License - v2.0 -- ../licenses/LICENSE-EPL-2.0.txt
  * Jakarta Injection -- org.glassfish.hk2.external-jakarta.inject-2.6.1.jar
 
 Public Domain (CC0) -- ../licenses/LICENSE-CC0.txt
- * Reactive Streams -- org.reactivestreams-reactive-streams-1.0.3.jar
+ * Reactive Streams -- org.reactivestreams-reactive-streams-1.0.4.jar
 
 Creative Commons Attribution License
  * Jcip -- ../licenses/LICENSE-jcip.txt
diff --git a/distribution/shell/src/assemble/LICENSE.bin.txt b/distribution/shell/src/assemble/LICENSE.bin.txt
@@ -398,8 +398,8 @@ The Apache Software License, Version 2.0
   * AirCompressor
      - aircompressor-2.0.3.jar
  * AsyncHttpClient
-    - async-http-client-2.12.4.jar
-    - async-http-client-netty-utils-2.12.4.jar
+    - async-http-client-2.14.5.jar
+    - async-http-client-netty-utils-2.14.5.jar
  * Jetty
     - jetty-client-9.4.58.v20250814.jar
     - jetty-http-9.4.58.v20250814.jar
@@ -462,7 +462,7 @@ Eclipse Public License - v2.0 -- ../licenses/LICENSE-EPL-2.0.txt
  * Jakarta Injection -- jakarta.inject-2.6.1.jar
 
 Public Domain (CC0) -- ../licenses/LICENSE-CC0.txt
- * Reactive Streams -- reactive-streams-1.0.3.jar
+ * Reactive Streams -- reactive-streams-1.0.4.jar
 
 Creative Commons Attribution License
  * Jcip -- ../licenses/LICENSE-jcip.txt
diff --git a/pom.xml b/pom.xml
@@ -258,7 +258,7 @@ flexible messaging model and an intuitive client API.</description>
     <prometheus-jmx.version>0.16.1</prometheus-jmx.version>
     <confluent.version>7.9.2</confluent.version>
     <aircompressor.version>2.0.3</aircompressor.version>
-    <asynchttpclient.version>2.12.4</asynchttpclient.version>
+    <asynchttpclient.version>2.14.5</asynchttpclient.version>
     <commons-lang3.version>3.19.0</commons-lang3.version>
     <commons-io.version>2.21.0</commons-io.version>
     <commons-codec.version>1.20.0</commons-codec.version>
diff --git a/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnectorTest.java b/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/http/AsyncHttpConnectorTest.java
@@ -19,7 +19,9 @@
 package org.apache.pulsar.client.admin.internal.http;
 
 import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
+import static com.github.tomakehurst.wiremock.client.WireMock.equalTo;
 import static com.github.tomakehurst.wiremock.client.WireMock.get;
+import static com.github.tomakehurst.wiremock.client.WireMock.getRequestedFor;
 import static com.github.tomakehurst.wiremock.client.WireMock.post;
 import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo;
 import static org.testng.Assert.assertEquals;
@@ -282,6 +284,60 @@ private void doTestRedirect(String location) throws InterruptedException, Execut
         assertEquals(response.getResponseBody(), "OK");
     }
 
+    /**
+     * Locks in that AsyncHttpConnector forwards the {@code Authorization} header across a cross-origin
+     * HTTP redirect (different host:port — serverA → serverB). The admin connector disables AHC's
+     * built-in follow-redirect and runs its own redirect loop that copies the original request headers,
+     * so it must remain unaffected by the async-http-client &gt;= 2.14.5 {@code Authorization} stripping
+     * on cross-origin redirects (CVE-2026-40490 fix). Regressing {@code setFollowRedirect(true)} would
+     * break this test.
+     */
+    @Test
+    void testAuthorizationHeaderOnCrossOriginRedirect() throws ExecutionException, InterruptedException {
+        WireMockServer serverB = new WireMockServer(WireMockConfiguration.wireMockConfig().port(0));
+        serverB.start();
+        try {
+            server.stubFor(get(urlEqualTo("/admin/v2/clusters"))
+                    .willReturn(aResponse()
+                            .withStatus(307)
+                            .withHeader("Location",
+                                    "http://127.0.0.1:" + serverB.port() + "/admin/v2/clusters")));
+
+            serverB.stubFor(get(urlEqualTo("/admin/v2/clusters"))
+                    .atPriority(2)
+                    .willReturn(aResponse().withStatus(401).withBody("missing auth")));
+            serverB.stubFor(get(urlEqualTo("/admin/v2/clusters"))
+                    .atPriority(1)
+                    .withHeader("Authorization", equalTo("Bearer test-token"))
+                    .willReturn(aResponse()
+                            .withStatus(200)
+                            .withHeader("Content-Type", "application/json")
+                            .withBody("[\"test-cluster\"]")));
+
+            ClientConfigurationData conf = new ClientConfigurationData();
+            conf.setServiceUrl("http://127.0.0.1:" + server.port());
+
+            @Cleanup
+            AsyncHttpConnector connector = new AsyncHttpConnector(5000, 5000,
+                    5000, 0, conf, false);
+
+            Request request = new RequestBuilder("GET")
+                    .setUrl("http://127.0.0.1:" + server.port() + "/admin/v2/clusters")
+                    .addHeader("Authorization", "Bearer test-token")
+                    .build();
+
+            Response response = connector.executeRequest(request).get();
+
+            assertEquals(response.getStatusCode(), 200,
+                    "cross-origin redirect should forward Authorization and return the stubbed body");
+            assertEquals(response.getResponseBody(), "[\"test-cluster\"]");
+            serverB.verify(getRequestedFor(urlEqualTo("/admin/v2/clusters"))
+                    .withHeader("Authorization", equalTo("Bearer test-token")));
+        } finally {
+            serverB.stop();
+        }
+    }
+
     @Test
     void testRedirectWithBody() throws ExecutionException, InterruptedException {
         server.stubFor(post(urlEqualTo("/path1"))
diff --git a/pulsar-client/pom.xml b/pulsar-client/pom.xml
@@ -228,6 +228,12 @@
       <artifactId>fastutil</artifactId>
     </dependency>
 
+    <dependency>
+      <groupId>com.github.tomakehurst</groupId>
+      <artifactId>wiremock-jre8-standalone</artifactId>
+      <version>${wiremock.version}</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpClient.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpClient.java
@@ -52,6 +52,7 @@
 import org.asynchttpclient.DefaultAsyncHttpClient;
 import org.asynchttpclient.DefaultAsyncHttpClientConfig;
 import org.asynchttpclient.Request;
+import org.asynchttpclient.Response;
 import org.asynchttpclient.SslEngineFactory;
 import org.asynchttpclient.channel.DefaultKeepAliveStrategy;
 
@@ -79,7 +80,12 @@ protected HttpClient(ClientConfigurationData conf, EventLoopGroup eventLoopGroup
         DefaultAsyncHttpClientConfig.Builder confBuilder = new DefaultAsyncHttpClientConfig.Builder();
         confBuilder.setCookieStore(null);
         confBuilder.setUseProxyProperties(true);
-        confBuilder.setFollowRedirect(true);
+        // Follow redirects manually in executeGet(...) so we can re-invoke authentication per hop and
+        // carry the Authorization header across cross-origin redirects. async-http-client >= 2.14.5
+        // (CVE-2026-40490 fix) strips the Authorization header when it follows redirects itself; Pulsar
+        // HTTP lookups routinely redirect to another broker's httpUrl/httpUrlTls which is a different
+        // host/port, i.e. cross-origin.
+        confBuilder.setFollowRedirect(false);
         confBuilder.setMaxRedirects(conf.getMaxLookupRedirects());
         confBuilder.setConnectTimeout(DEFAULT_CONNECT_TIMEOUT_IN_SECONDS * 1000);
         confBuilder.setReadTimeout(DEFAULT_READ_TIMEOUT_IN_SECONDS * 1000);
@@ -156,7 +162,24 @@ public <T> CompletableFuture<T> get(String path, Class<T> clazz) {
         try {
             URI hostUri = serviceNameResolver.resolveHostUri();
             String requestUrl = new URL(hostUri.toURL(), path).toString();
-            String remoteHostName = hostUri.getHost();
+            executeGet(requestUrl, clientConf.getMaxLookupRedirects(), future, clazz);
+        } catch (Exception e) {
+            log.warn("[{}] Failed to initiate HTTP get request: {}", path, e.getMessage());
+            if (e instanceof PulsarClientException) {
+                future.completeExceptionally(e);
+            } else {
+                future.completeExceptionally(new PulsarClientException(e));
+            }
+        }
+
+        return future;
+    }
+
+    private <T> void executeGet(String requestUrl, int redirectsRemaining,
+                                CompletableFuture<T> future, Class<T> clazz) {
+        try {
+            URI currentUri = URI.create(requestUrl);
+            String remoteHostName = currentUri.getHost();
             AuthenticationDataProvider authData = authentication.getAuthData(remoteHostName);
 
             CompletableFuture<Map<String, String>>  authFuture = new CompletableFuture<>();
@@ -207,11 +230,17 @@ public <T> CompletableFuture<T> get(String path, Class<T> clazz) {
                         return;
                     }
 
+                    int statusCode = response2.getStatusCode();
+                    if (isRedirectStatusCode(statusCode)) {
+                        handleRedirect(requestUrl, currentUri, response2, redirectsRemaining, future, clazz);
+                        return;
+                    }
+
                     // request not success
-                    if (response2.getStatusCode() != HttpURLConnection.HTTP_OK) {
+                    if (statusCode != HttpURLConnection.HTTP_OK) {
                         log.warn("[{}] HTTP get request failed: {}", requestUrl, response2.getStatusText());
                         Exception e;
-                        if (response2.getStatusCode() == HttpURLConnection.HTTP_NOT_FOUND) {
+                        if (statusCode == HttpURLConnection.HTTP_NOT_FOUND) {
                             e = new NotFoundException("Not found: " + response2.getStatusText());
                         } else {
                             e = new PulsarClientException("HTTP get request failed: " + response2.getStatusText());
@@ -231,15 +260,46 @@ public <T> CompletableFuture<T> get(String path, Class<T> clazz) {
                 });
             });
         } catch (Exception e) {
-            log.warn("[{}]PulsarClientImpl: {}", path, e.getMessage());
+            log.warn("[{}] HTTP request setup failed: {}", requestUrl, e.getMessage());
             if (e instanceof PulsarClientException) {
                 future.completeExceptionally(e);
             } else {
                 future.completeExceptionally(new PulsarClientException(e));
             }
         }
+    }
 
-        return future;
+    private <T> void handleRedirect(String requestUrl, URI currentUri, Response response,
+                                    int redirectsRemaining, CompletableFuture<T> future, Class<T> clazz) {
+        String location = response.getHeader("Location");
+        if (location == null || location.isEmpty()) {
+            future.completeExceptionally(new PulsarClientException(
+                    "HTTP redirect " + response.getStatusCode() + " without Location header: " + requestUrl));
+            return;
+        }
+        if (redirectsRemaining <= 0) {
+            future.completeExceptionally(new PulsarClientException(
+                    "Maximum redirects exceeded (" + clientConf.getMaxLookupRedirects()
+                            + ") while following HTTP redirect for " + requestUrl));
+            return;
+        }
+        String newUrl;
+        try {
+            newUrl = currentUri.resolve(location).toString();
+        } catch (Exception e) {
+            future.completeExceptionally(new PulsarClientException(
+                    "Invalid redirect Location \"" + location + "\" for " + requestUrl));
+            return;
+        }
+        executeGet(newUrl, redirectsRemaining - 1, future, clazz);
+    }
+
+    private static boolean isRedirectStatusCode(int statusCode) {
+        return statusCode == HttpURLConnection.HTTP_MOVED_PERM   // 301
+                || statusCode == HttpURLConnection.HTTP_MOVED_TEMP   // 302
+                || statusCode == HttpURLConnection.HTTP_SEE_OTHER    // 303
+                || statusCode == 307                                 // Temporary Redirect
+                || statusCode == 308;                                // Permanent Redirect
     }
 
     protected PulsarSslConfiguration buildSslConfiguration(ClientConfigurationData config, String host)
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/HttpClientTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/HttpClientTest.java
@@ -0,0 +1,137 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.pulsar.client.impl;
+
+import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
+import static com.github.tomakehurst.wiremock.client.WireMock.equalTo;
+import static com.github.tomakehurst.wiremock.client.WireMock.get;
+import static com.github.tomakehurst.wiremock.client.WireMock.getRequestedFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.urlPathMatching;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import com.github.tomakehurst.wiremock.WireMockServer;
+import com.github.tomakehurst.wiremock.core.WireMockConfiguration;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.util.concurrent.TimeUnit;
+import org.apache.pulsar.client.api.AuthenticationFactory;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.common.lookup.data.LookupData;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+/**
+ * Verifies that {@link HttpClient} carries the {@code Authorization} header across cross-origin HTTP redirects.
+ *
+ * <p>Pulsar's HTTP lookup endpoint returns {@code 307 Temporary Redirect} to whichever broker owns the bundle for a
+ * topic. The redirect target is that broker's {@code httpUrl}/{@code httpUrlTls}, i.e. typically a different host or
+ * port from the original request. Auth plugins ({@code AuthenticationToken}, {@code AuthenticationBasic},
+ * {@code AuthenticationOAuth2}, {@code AuthenticationAthenz}) inject the {@code Authorization} header — that header
+ * must reach the redirect target for lookup to succeed.
+ *
+ * <p>async-http-client 2.14.5 strips {@code Authorization} on cross-origin redirects when its built-in follow-redirect
+ * is enabled (CVE-2026-40490 fix). This test drives two WireMock servers on different ports to exercise that path.
+ */
+public class HttpClientTest {
+
+    private static final String LOOKUP_PATH = "/lookup/v2/topic/persistent/public/default/test-topic";
+    private static final String EXPECTED_BODY = "{\"brokerUrl\":\"pulsar://broker-b:6650\","
+            + "\"brokerUrlTls\":\"pulsar+ssl://broker-b:6651\","
+            + "\"httpUrl\":\"http://broker-b:8080\","
+            + "\"httpUrlTls\":\"https://broker-b:8443\","
+            + "\"nativeUrl\":\"pulsar://broker-b:6650\"}";
+
+    private WireMockServer serverA;
+    private WireMockServer serverB;
+    private EventLoopGroup eventLoopGroup;
+
+    @BeforeClass(alwaysRun = true)
+    void beforeClass() {
+        eventLoopGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("HttpClientTest"));
+    }
+
+    @AfterClass(alwaysRun = true)
+    void afterClass() {
+        if (eventLoopGroup != null) {
+            eventLoopGroup.shutdownGracefully();
+        }
+    }
+
+    @BeforeMethod(alwaysRun = true)
+    void beforeMethod() {
+        serverA = new WireMockServer(WireMockConfiguration.wireMockConfig().port(0));
+        serverB = new WireMockServer(WireMockConfiguration.wireMockConfig().port(0));
+        serverA.start();
+        serverB.start();
+    }
+
+    @AfterMethod(alwaysRun = true)
+    void afterMethod() {
+        if (serverA != null) {
+            serverA.stop();
+        }
+        if (serverB != null) {
+            serverB.stop();
+        }
+    }
+
+    @Test
+    public void testCrossOriginRedirectCarriesAuthorizationHeader() throws Exception {
+        // serverA (origin host:port) returns 307 Temporary Redirect to serverB (different port -> cross-origin).
+        serverA.stubFor(get(urlPathMatching(LOOKUP_PATH))
+                .willReturn(aResponse()
+                        .withStatus(307)
+                        .withHeader("Location",
+                                "http://127.0.0.1:" + serverB.port() + LOOKUP_PATH)));
+
+        // serverB only responds 200 when the Authorization header is present with the expected token.
+        // Priorities: lower = higher priority; the specific-Authorization stub must be checked first.
+        serverB.stubFor(get(urlPathMatching(LOOKUP_PATH))
+                .atPriority(2)
+                .willReturn(aResponse().withStatus(401).withBody("missing auth")));
+        serverB.stubFor(get(urlPathMatching(LOOKUP_PATH))
+                .atPriority(1)
+                .withHeader("Authorization", equalTo("Bearer test-token"))
+                .willReturn(aResponse()
+                        .withStatus(200)
+                        .withHeader("Content-Type", "application/json")
+                        .withBody(EXPECTED_BODY)));
+
+        ClientConfigurationData conf = new ClientConfigurationData();
+        conf.setServiceUrl("http://127.0.0.1:" + serverA.port());
+        conf.setAuthentication(AuthenticationFactory.token("test-token"));
+
+        try (HttpClient httpClient = new HttpClient(conf, eventLoopGroup)) {
+            LookupData result = httpClient.get(LOOKUP_PATH, LookupData.class)
+                    .get(30, TimeUnit.SECONDS);
+
+            assertNotNull(result, "Expected lookup payload after cross-origin redirect");
+            assertEquals(result.getBrokerUrl(), "pulsar://broker-b:6650");
+            assertEquals(result.getHttpUrl(), "http://broker-b:8080");
+
+            // Lock the invariant: the final hop on serverB must carry the Authorization header.
+            serverB.verify(getRequestedFor(urlPathMatching(LOOKUP_PATH))
+                    .withHeader("Authorization", equalTo("Bearer test-token")));
+        }
+    }
+}
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/HttpLookupServiceTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/HttpLookupServiceTest.java
