This is an automated email from the ASF dual-hosted git repository.
oscerd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 49ca42792152 CAMEL-23876: camel-a2a - pin push notification webhooks
to the validated address (#25004)
49ca42792152 is described below
commit 49ca42792152c7e0987a8f3dc9f8d60c758ffe59
Author: Andrea Cosentino <[email protected]>
AuthorDate: Wed Jul 22 15:02:40 2026 +0200
CAMEL-23876: camel-a2a - pin push notification webhooks to the validated
address (#25004)
Deliver push notifications with Apache HttpClient 5, whose HttpHost carries
both an explicit InetAddress and the hostname: the connection is opened to the
address that passed SSRF validation, while the Host header, TLS SNI and
certificate hostname verification still use the hostname. Validation and
resolution now run on every attempt.
Co-authored-by: Claude Fable 5 <[email protected]>
---
components/camel-ai/camel-a2a/pom.xml | 6 +
.../apache/camel/component/a2a/A2AEndpoint.java | 31 ++++-
.../a2a/push/PushNotificationDispatcher.java | 139 +++++++++++++--------
.../component/a2a/util/WebhookUrlValidator.java | 20 ++-
.../a2a/push/PushNotificationDispatcherTest.java | 42 +++++--
.../push/PushNotificationTargetPinningTest.java | 68 ++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_22.adoc | 20 +++
7 files changed, 261 insertions(+), 65 deletions(-)
diff --git a/components/camel-ai/camel-a2a/pom.xml
b/components/camel-ai/camel-a2a/pom.xml
index 96a944a85b67..e4171df1f8e2 100644
--- a/components/camel-ai/camel-a2a/pom.xml
+++ b/components/camel-ai/camel-a2a/pom.xml
@@ -64,6 +64,12 @@
<groupId>com.fasterxml.jackson.datatype</groupId>
<artifactId>jackson-datatype-jsr310</artifactId>
</dependency>
+ <!-- used to dispatch push notifications to a connection pinned to the
validated webhook address -->
+ <dependency>
+ <groupId>org.apache.httpcomponents.client5</groupId>
+ <artifactId>httpclient5</artifactId>
+ <version>${httpclient-version}</version>
+ </dependency>
<!-- Test dependencies -->
<dependency>
diff --git
a/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/A2AEndpoint.java
b/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/A2AEndpoint.java
index dc5f8153d934..a0f8cca59ddc 100644
---
a/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/A2AEndpoint.java
+++
b/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/A2AEndpoint.java
@@ -53,6 +53,10 @@ import org.apache.camel.spi.UriParam;
import org.apache.camel.spi.UriPath;
import org.apache.camel.support.DefaultEndpoint;
import org.apache.camel.support.service.ServiceHelper;
+import org.apache.hc.client5.http.config.RequestConfig;
+import org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient;
+import org.apache.hc.client5.http.impl.async.HttpAsyncClients;
+import org.apache.hc.core5.util.Timeout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -86,6 +90,7 @@ public class A2AEndpoint extends DefaultEndpoint {
private ScheduledExecutorService pushDispatcherExecutor;
private ExecutorService httpClientExecutor;
private HttpClient httpClient;
+ private CloseableHttpAsyncClient pushWebhookHttpClient;
private Map<String, A2AExtensionHandler> extensionHandlers = Map.of();
private boolean producerCreated;
private boolean consumerCreated;
@@ -180,8 +185,10 @@ public class A2AEndpoint extends DefaultEndpoint {
if (initializeConsumerServices) {
pushDispatcherExecutor =
getCamelContext().getExecutorServiceManager()
.newScheduledThreadPool(this, "A2APushDispatcher", 4);
+ pushWebhookHttpClient = createPushWebhookHttpClient();
+ pushWebhookHttpClient.start();
pushDispatcher = new PushNotificationDispatcher(
- createHttpClient(HttpClient.Redirect.NEVER), taskStore,
+ pushWebhookHttpClient, taskStore,
configuration.getPushRetryAttempts(),
configuration.getPushRetryBackoffMs(),
pushDispatcherExecutor,
@@ -251,6 +258,20 @@ public class A2AEndpoint extends DefaultEndpoint {
.build();
}
+ /**
+ * Creates the client used to deliver push notifications to
caller-registered webhooks. Redirects are never
+ * followed: a redirect would be resolved and connected to by the client
itself, escaping the SSRF validation
+ * applied to the registered URL. The dispatcher pins each request to the
address it validated.
+ */
+ private CloseableHttpAsyncClient createPushWebhookHttpClient() {
+ return HttpAsyncClients.custom()
+ .disableRedirectHandling()
+ .setDefaultRequestConfig(RequestConfig.custom()
+
.setConnectTimeout(Timeout.ofMilliseconds(configuration.getConnectTimeout()))
+ .build())
+ .build();
+ }
+
private void cleanupEndpointResources() {
if (taskStoreOwned && taskStore != null) {
try {
@@ -263,6 +284,14 @@ public class A2AEndpoint extends DefaultEndpoint {
pushDispatcher.shutdown();
}
pushDispatcher = null;
+ if (pushWebhookHttpClient != null) {
+ try {
+ pushWebhookHttpClient.close();
+ } catch (Exception e) {
+ LOG.debug("Error closing A2A push webhook HTTP client: {}",
e.getMessage());
+ }
+ pushWebhookHttpClient = null;
+ }
if (pushDispatcherExecutor != null) {
getCamelContext().getExecutorServiceManager().shutdownNow(pushDispatcherExecutor);
pushDispatcherExecutor = null;
diff --git
a/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/push/PushNotificationDispatcher.java
b/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/push/PushNotificationDispatcher.java
index 344dc73c742b..711714a42361 100644
---
a/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/push/PushNotificationDispatcher.java
+++
b/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/push/PushNotificationDispatcher.java
@@ -16,12 +16,8 @@
*/
package org.apache.camel.component.a2a.push;
-import java.io.IOException;
-import java.io.InputStream;
+import java.net.InetAddress;
import java.net.URI;
-import java.net.http.HttpClient;
-import java.net.http.HttpRequest;
-import java.net.http.HttpResponse;
import java.time.Duration;
import java.util.List;
import java.util.Set;
@@ -44,6 +40,17 @@ import
org.apache.camel.component.a2a.model.TaskPushNotificationConfig;
import org.apache.camel.component.a2a.state.A2ATaskStore;
import org.apache.camel.component.a2a.util.A2AJsonMapper;
import org.apache.camel.component.a2a.util.WebhookUrlValidator;
+import org.apache.hc.client5.http.async.methods.SimpleHttpRequest;
+import org.apache.hc.client5.http.async.methods.SimpleHttpResponse;
+import org.apache.hc.client5.http.async.methods.SimpleRequestBuilder;
+import org.apache.hc.client5.http.async.methods.SimpleRequestProducer;
+import org.apache.hc.client5.http.async.methods.SimpleResponseConsumer;
+import org.apache.hc.client5.http.config.RequestConfig;
+import org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient;
+import org.apache.hc.core5.concurrent.FutureCallback;
+import org.apache.hc.core5.http.ContentType;
+import org.apache.hc.core5.http.HttpHost;
+import org.apache.hc.core5.util.Timeout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -62,7 +69,7 @@ public class PushNotificationDispatcher {
private static final long MAX_RETRY_DELAY_MS = TimeUnit.HOURS.toMillis(1);
private static final int MAX_PENDING_WORK = 1024;
- private final HttpClient httpClient;
+ private final CloseableHttpAsyncClient httpClient;
private final A2ATaskStore store;
private final int maxRetries;
private final long initialBackoffMs;
@@ -73,7 +80,7 @@ public class PushNotificationDispatcher {
private final Set<CompletableFuture<?>> inFlightRequests =
ConcurrentHashMap.newKeySet();
private final Set<ScheduledFuture<?>> scheduledRetries =
ConcurrentHashMap.newKeySet();
- public PushNotificationDispatcher(HttpClient httpClient, A2ATaskStore
store,
+ public PushNotificationDispatcher(CloseableHttpAsyncClient httpClient,
A2ATaskStore store,
int maxRetries, long initialBackoffMs,
ScheduledExecutorService executor,
boolean allowLocalWebhookUrls) {
@@ -133,51 +140,90 @@ public class PushNotificationDispatcher {
}
private void dispatchToWebhook(TaskPushNotificationConfig config, byte[]
body) {
- if (closed.get()) {
- return;
- }
- try {
- WebhookUrlValidator.validate(config.getUrl(),
allowLocalWebhookUrls);
- } catch (IllegalArgumentException e) {
- LOG.warn("Skipping invalid push notification webhook for
taskId={}: {}", config.getTaskId(), e.getMessage());
- return;
- }
+ sendWithRetryAttempt(config, body, 0);
+ }
- HttpRequest.Builder requestBuilder = HttpRequest.newBuilder()
- .uri(URI.create(config.getUrl()))
- .header("Content-Type", A2AConstants.CONTENT_TYPE)
- .timeout(REQUEST_TIMEOUT)
- .POST(HttpRequest.BodyPublishers.ofByteArray(body));
+ /**
+ * Builds a request target pinned to the address the webhook URL was just
validated against.
+ * <p>
+ * The host is resolved once, by the validator, and the resulting address
is carried on the {@link HttpHost} so the
+ * connection is opened to exactly that address. The original hostname is
kept on the same {@link HttpHost}, so the
+ * {@code Host} header, TLS SNI and certificate hostname verification
still use the hostname. Without this the HTTP
+ * client would resolve the hostname again at connection time and could
reach a different address than the one that
+ * passed the SSRF checks (DNS rebinding).
+ */
+ static HttpHost pinnedTarget(URI uri, InetAddress validatedAddress) {
+ String scheme = uri.getScheme();
+ int port = uri.getPort();
+ if (port == -1) {
+ port = "https".equalsIgnoreCase(scheme) ? 443 : 80;
+ }
+ return new HttpHost(scheme, validatedAddress, uri.getHost(), port);
+ }
+ private SimpleHttpRequest buildRequest(URI uri, TaskPushNotificationConfig
config, byte[] body) {
+ SimpleRequestBuilder requestBuilder = SimpleRequestBuilder.post(uri)
+ .setBody(body, ContentType.parse(A2AConstants.CONTENT_TYPE))
+ .setRequestConfig(RequestConfig.custom()
+
.setResponseTimeout(Timeout.ofMilliseconds(REQUEST_TIMEOUT.toMillis()))
+ .build());
applyAuth(requestBuilder, config);
- HttpRequest request = requestBuilder.build();
-
- sendWithRetry(request, config);
+ return requestBuilder.build();
}
- private void applyAuth(HttpRequest.Builder builder,
TaskPushNotificationConfig config) {
+ private void applyAuth(SimpleRequestBuilder builder,
TaskPushNotificationConfig config) {
AuthenticationInfo auth = config.getAuthentication();
if (auth != null && auth.getScheme() != null && auth.getCredentials()
!= null) {
- builder.header("Authorization", auth.getScheme() + " " +
auth.getCredentials());
+ builder.setHeader("Authorization", auth.getScheme() + " " +
auth.getCredentials());
} else if (config.getToken() != null) {
- builder.header("Authorization", "Bearer " + config.getToken());
+ builder.setHeader("Authorization", "Bearer " + config.getToken());
}
}
- private void sendWithRetry(HttpRequest request, TaskPushNotificationConfig
config) {
- sendWithRetryAttempt(request, config, 0);
- }
-
- private void sendWithRetryAttempt(HttpRequest request,
TaskPushNotificationConfig config, int attempt) {
+ private void sendWithRetryAttempt(TaskPushNotificationConfig config,
byte[] body, int attempt) {
if (closed.get()) {
return;
}
+
+ // Re-validate and re-resolve on every attempt: a config that was safe
when first dispatched can be
+ // rebound between retries, which may be up to an hour apart.
+ URI uri;
+ InetAddress validatedAddress;
+ try {
+ validatedAddress =
WebhookUrlValidator.validateAndResolve(config.getUrl(), allowLocalWebhookUrls);
+ uri = URI.create(config.getUrl());
+ } catch (IllegalArgumentException e) {
+ LOG.warn("Skipping invalid push notification webhook for
taskId={}: {}", config.getTaskId(), e.getMessage());
+ return;
+ }
+
if (!tryAdmitWork(config, "HTTP request")) {
return;
}
- CompletableFuture<HttpResponse<InputStream>> responseFuture;
+
+ CompletableFuture<SimpleHttpResponse> responseFuture = new
CompletableFuture<>();
try {
- responseFuture = httpClient.sendAsync(request,
HttpResponse.BodyHandlers.ofInputStream());
+ httpClient.execute(
+ pinnedTarget(uri, validatedAddress),
+ SimpleRequestProducer.create(buildRequest(uri, config,
body)),
+ SimpleResponseConsumer.create(),
+ null, null,
+ new FutureCallback<>() {
+ @Override
+ public void completed(SimpleHttpResponse response) {
+ responseFuture.complete(response);
+ }
+
+ @Override
+ public void failed(Exception e) {
+ responseFuture.completeExceptionally(e);
+ }
+
+ @Override
+ public void cancelled() {
+ responseFuture.cancel(false);
+ }
+ });
} catch (RuntimeException e) {
releaseWork();
throw e;
@@ -186,14 +232,10 @@ public class PushNotificationDispatcher {
try {
track(responseFuture.whenCompleteAsync((response, failure) -> {
if (closed.get()) {
- if (response != null) {
- closeResponseBody(response.body(), config);
- }
return;
}
- boolean retry = false;
+ boolean retry;
if (failure == null) {
- closeResponseBody(response.body(), config);
retry = shouldRetry(response, config, attempt);
} else {
Throwable cause = failure instanceof CompletionException
&& failure.getCause() != null
@@ -204,7 +246,7 @@ public class PushNotificationDispatcher {
retry = true;
}
if (retry && attempt < maxRetries) {
- scheduleRetry(request, config, attempt);
+ scheduleRetry(config, body, attempt);
} else if (retry) {
LOG.error("Push notification to {} exhausted all {}
retries for taskId={}",
config.getUrl(), maxRetries + 1,
config.getTaskId());
@@ -216,7 +258,7 @@ public class PushNotificationDispatcher {
}
}
- private void scheduleRetry(HttpRequest request, TaskPushNotificationConfig
config, int attempt) {
+ private void scheduleRetry(TaskPushNotificationConfig config, byte[] body,
int attempt) {
if (closed.get()) {
return;
}
@@ -229,7 +271,7 @@ public class PushNotificationDispatcher {
ScheduledFuture<?> scheduled = executor.schedule(() -> {
scheduledRetries.remove(scheduledRef.get());
releaseWork();
- sendWithRetryAttempt(request, config, attempt + 1);
+ sendWithRetryAttempt(config, body, attempt + 1);
}, delay, TimeUnit.MILLISECONDS);
scheduledRef.set(scheduled);
scheduledRetries.add(scheduled);
@@ -287,8 +329,8 @@ public class PushNotificationDispatcher {
pendingWork.updateAndGet(current -> current > 0 ? current - 1 : 0);
}
- private boolean shouldRetry(HttpResponse<InputStream> response,
TaskPushNotificationConfig config, int attempt) {
- int status = response.statusCode();
+ private boolean shouldRetry(SimpleHttpResponse response,
TaskPushNotificationConfig config, int attempt) {
+ int status = response.getCode();
if (status >= 200 && status < 300) {
LOG.debug("Push notification delivered to {} for taskId={}",
@@ -329,13 +371,4 @@ public class PushNotificationDispatcher {
executor.shutdownNow();
}
}
-
- private static void closeResponseBody(InputStream body,
TaskPushNotificationConfig config) {
- try {
- body.close();
- } catch (IOException e) {
- LOG.debug("Failed to close push notification response body for
taskId={}: {}",
- config.getTaskId(), e.getMessage());
- }
- }
}
diff --git
a/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/util/WebhookUrlValidator.java
b/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/util/WebhookUrlValidator.java
index 36f8a1b116fe..cd8a41787584 100644
---
a/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/util/WebhookUrlValidator.java
+++
b/components/camel-ai/camel-a2a/src/main/java/org/apache/camel/component/a2a/util/WebhookUrlValidator.java
@@ -46,6 +46,22 @@ public final class WebhookUrlValidator {
* @throws IllegalArgumentException if the URL is invalid or unsafe
*/
public static void validate(String url, boolean allowLocal) {
+ validateAndResolve(url, allowLocal);
+ }
+
+ /**
+ * Validates a webhook URL for SSRF protection and returns the address the
host resolved to during validation.
+ * <p>
+ * Callers that go on to open a connection should connect to the returned
address rather than letting the HTTP
+ * client resolve the hostname again, otherwise the address that was
validated and the address actually connected to
+ * can differ for the same hostname (DNS rebinding), defeating the checks
performed here.
+ *
+ * @param url the URL to validate
+ * @param allowLocal whether to permit loopback/localhost
addresses (dev mode)
+ * @return the validated address the host
resolved to
+ * @throws IllegalArgumentException if the URL is invalid or unsafe
+ */
+ public static InetAddress validateAndResolve(String url, boolean
allowLocal) {
if (url == null || url.isBlank()) {
throw new IllegalArgumentException("Webhook URL must not be null
or empty");
}
@@ -92,7 +108,7 @@ public final class WebhookUrlValidator {
+ ". Set
allowLocalWebhookUrls=true for local development.");
}
// Loopback allowed — skip remaining network checks, permit HTTP
- return;
+ return address;
}
// Non-loopback: require HTTPS
@@ -112,6 +128,8 @@ public final class WebhookUrlValidator {
throw new IllegalArgumentException(
"Webhook URL must not point to a site-local/private
address (SSRF protection): " + host);
}
+
+ return address;
}
private static boolean isPrivateIpv6(String host) {
diff --git
a/components/camel-ai/camel-a2a/src/test/java/org/apache/camel/component/a2a/push/PushNotificationDispatcherTest.java
b/components/camel-ai/camel-a2a/src/test/java/org/apache/camel/component/a2a/push/PushNotificationDispatcherTest.java
index d6715f72c67c..5f629ed8d7d7 100644
---
a/components/camel-ai/camel-a2a/src/test/java/org/apache/camel/component/a2a/push/PushNotificationDispatcherTest.java
+++
b/components/camel-ai/camel-a2a/src/test/java/org/apache/camel/component/a2a/push/PushNotificationDispatcherTest.java
@@ -18,7 +18,8 @@ package org.apache.camel.component.a2a.push;
import java.lang.reflect.Field;
import java.net.InetSocketAddress;
-import java.net.http.HttpClient;
+import java.util.ArrayList;
+import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
@@ -34,6 +35,8 @@ import org.apache.camel.component.a2a.model.TaskState;
import org.apache.camel.component.a2a.model.TaskStatus;
import org.apache.camel.component.a2a.model.TaskStatusUpdateEvent;
import org.apache.camel.component.a2a.state.InMemoryTaskStore;
+import org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient;
+import org.apache.hc.client5.http.impl.async.HttpAsyncClients;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -49,6 +52,17 @@ class PushNotificationDispatcherTest {
private InMemoryTaskStore store;
private PushNotificationDispatcher dispatcher;
private ScheduledExecutorService executor;
+ private final List<CloseableHttpAsyncClient> httpClients = new
ArrayList<>();
+
+ /**
+ * Creates and starts an async HTTP client, tracked so it is closed after
the test.
+ */
+ private CloseableHttpAsyncClient startedClient() {
+ CloseableHttpAsyncClient client =
HttpAsyncClients.custom().disableRedirectHandling().build();
+ client.start();
+ httpClients.add(client);
+ return client;
+ }
@BeforeEach
void setUp() throws Exception {
@@ -70,6 +84,14 @@ class PushNotificationDispatcherTest {
if (mockServer != null) {
mockServer.stop(0);
}
+ for (CloseableHttpAsyncClient client : httpClients) {
+ try {
+ client.close();
+ } catch (Exception e) {
+ // ignore on teardown
+ }
+ }
+ httpClients.clear();
}
@Test
@@ -87,7 +109,7 @@ class PushNotificationDispatcherTest {
mockServer.start();
dispatcher = new PushNotificationDispatcher(
- HttpClient.newHttpClient(), store, 0, 1000, executor, true);
+ startedClient(), store, 0, 1000, executor, true);
TaskPushNotificationConfig config = new TaskPushNotificationConfig();
config.setUrl("http://localhost:"; + port + "/webhook");
@@ -122,7 +144,7 @@ class PushNotificationDispatcherTest {
mockServer.start();
dispatcher = new PushNotificationDispatcher(
- HttpClient.newHttpClient(), store, 3, 50, executor, true);
+ startedClient(), store, 3, 50, executor, true);
TaskPushNotificationConfig config = new TaskPushNotificationConfig();
config.setUrl("http://localhost:"; + port + "/webhook");
@@ -149,7 +171,7 @@ class PushNotificationDispatcherTest {
mockServer.start();
dispatcher = new PushNotificationDispatcher(
- HttpClient.newHttpClient(), store, 0, 1000, executor, true);
+ startedClient(), store, 0, 1000, executor, true);
TaskPushNotificationConfig config = new TaskPushNotificationConfig();
config.setUrl("http://localhost:"; + port + "/webhook");
@@ -179,7 +201,7 @@ class PushNotificationDispatcherTest {
mockServer.start();
dispatcher = new PushNotificationDispatcher(
- HttpClient.newHttpClient(), store, 3, 50, executor, true);
+ startedClient(), store, 3, 50, executor, true);
TaskPushNotificationConfig config = new TaskPushNotificationConfig();
config.setUrl("http://localhost:"; + port + "/webhook");
@@ -206,7 +228,7 @@ class PushNotificationDispatcherTest {
mockServer.start();
dispatcher = new PushNotificationDispatcher(
- HttpClient.newHttpClient(), store, 1, 5000, executor, true);
+ startedClient(), store, 1, 5000, executor, true);
TaskPushNotificationConfig config = new TaskPushNotificationConfig();
config.setUrl("http://localhost:"; + port + "/webhook");
@@ -243,7 +265,7 @@ class PushNotificationDispatcherTest {
mockServer.start();
dispatcher = new PushNotificationDispatcher(
- HttpClient.newHttpClient(), store, 0, 1000, executor, true);
+ startedClient(), store, 0, 1000, executor, true);
for (int i = 1; i <= 3; i++) {
TaskPushNotificationConfig config = new
TaskPushNotificationConfig();
@@ -263,7 +285,7 @@ class PushNotificationDispatcherTest {
@Test
void dispatchSkipsWhenNoConfigs() {
dispatcher = new PushNotificationDispatcher(
- HttpClient.newHttpClient(), store, 0, 1000, executor, true);
+ startedClient(), store, 0, 1000, executor, true);
TaskStatusUpdateEvent event = TaskStatusUpdateEvent.builder()
.taskId("task-1")
@@ -275,7 +297,7 @@ class PushNotificationDispatcherTest {
@Test
void rejectsTightRetryLoopConfiguration() {
assertThatThrownBy(() -> new PushNotificationDispatcher(
- HttpClient.newHttpClient(), store, 1, 0, executor, true))
+ startedClient(), store, 1, 0, executor, true))
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("initialBackoffMs");
}
@@ -293,7 +315,7 @@ class PushNotificationDispatcherTest {
mockServer.start();
dispatcher = new PushNotificationDispatcher(
- HttpClient.newHttpClient(), store, 0, 1000, executor, true);
+ startedClient(), store, 0, 1000, executor, true);
TaskPushNotificationConfig config = new TaskPushNotificationConfig();
config.setUrl("http://localhost:"; + port + "/webhook");
diff --git
a/components/camel-ai/camel-a2a/src/test/java/org/apache/camel/component/a2a/push/PushNotificationTargetPinningTest.java
b/components/camel-ai/camel-a2a/src/test/java/org/apache/camel/component/a2a/push/PushNotificationTargetPinningTest.java
new file mode 100644
index 000000000000..a9772112af71
--- /dev/null
+++
b/components/camel-ai/camel-a2a/src/test/java/org/apache/camel/component/a2a/push/PushNotificationTargetPinningTest.java
@@ -0,0 +1,68 @@
+/*
+ * 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.camel.component.a2a.push;
+
+import java.net.InetAddress;
+import java.net.URI;
+
+import org.apache.hc.core5.http.HttpHost;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Verifies that a push notification request is targeted at the address the
webhook URL was validated against, while
+ * still carrying the original hostname so the Host header, TLS SNI and
certificate verification are unaffected.
+ */
+class PushNotificationTargetPinningTest {
+
+ @Test
+ void targetCarriesValidatedAddressAndOriginalHostname() throws Exception {
+ InetAddress validated = InetAddress.getByName("203.0.113.10");
+
+ HttpHost target = PushNotificationDispatcher.pinnedTarget(
+ URI.create("https://webhook.example.com/events";), validated);
+
+ // the connection is opened to the address that passed the SSRF
checks, not to a fresh DNS lookup
+ assertThat(target.getAddress()).isEqualTo(validated);
+ // the hostname is preserved for Host header / TLS SNI / certificate
hostname verification
+ assertThat(target.getHostName()).isEqualTo("webhook.example.com");
+ assertThat(target.getSchemeName()).isEqualTo("https");
+ assertThat(target.getPort()).isEqualTo(443);
+ }
+
+ @Test
+ void targetUsesDefaultPortPerScheme() throws Exception {
+ InetAddress validated = InetAddress.getByName("203.0.113.11");
+
+
assertThat(PushNotificationDispatcher.pinnedTarget(URI.create("https://host.example/x";),
validated).getPort())
+ .isEqualTo(443);
+
assertThat(PushNotificationDispatcher.pinnedTarget(URI.create("http://host.example/x";),
validated).getPort())
+ .isEqualTo(80);
+ }
+
+ @Test
+ void targetHonoursExplicitPort() throws Exception {
+ InetAddress validated = InetAddress.getByName("203.0.113.12");
+
+ HttpHost target = PushNotificationDispatcher.pinnedTarget(
+ URI.create("https://host.example:8443/x";), validated);
+
+ assertThat(target.getPort()).isEqualTo(8443);
+ assertThat(target.getAddress()).isEqualTo(validated);
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
index 1d83c20e43a7..6d822f0742dc 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
@@ -735,6 +735,26 @@ without being secrets — `adjustAuthorization`,
`jwtAuthorizationType`, `proxyA
have their values shown as `xxxxxx` in sanitized URIs. This is a cosmetic
change to sanitized output only and
does not affect the actual endpoint configuration.
+=== camel-a2a - push notification webhooks pinned to the validated address
+
+Push notifications are now delivered with Apache HttpClient 5 instead of
`java.net.http.HttpClient`, so the request
+can be pinned to the address the webhook URL was validated against.
+
+Previously `WebhookUrlValidator` resolved the webhook host and applied the
SSRF checks, and the HTTP client then
+resolved the same hostname again when it opened the connection. The two
lookups are independent, so the address that
+was validated and the address actually connected to could differ (DNS
rebinding). The connection is now opened to the
+address that passed validation, while the hostname is still used for the
`Host` header, TLS SNI and certificate
+hostname verification. Each retry attempt re-validates and re-resolves, so a
config that was safe when first
+dispatched cannot be rebound between retries (which may be up to an hour
apart).
+
+`camel-a2a` therefore has a new dependency on
`org.apache.httpcomponents.client5:httpclient5`.
+
+The `PushNotificationDispatcher` constructor now takes an
+`org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient` instead of a
`java.net.http.HttpClient`. This only
+affects code constructing the dispatcher directly; the client must be started
before use. Endpoint behaviour is
+unchanged — redirects were already never followed for push webhooks, and still
are not, since a redirect would be
+resolved by the client and escape the validation applied to the registered URL.
+
=== camel-core - Multicast UseOriginalAggregationStrategy fix
The Multicast EIP now correctly honors `UseOriginalAggregationStrategy`,
consistent with the Splitter
