This is an automated email from the ASF dual-hosted git repository.
Croway 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 56aa5f4970a6 CAMEL-25917: camel-undertow - a WebSocket connection
accepted while only a producer served the path is not treated as having passed
a later consumer's custom handlers (#28237)
56aa5f4970a6 is described below
commit 56aa5f4970a6fb7126275fc02b5bb9694dd184d1
Author: Federico Mariani <[email protected]>
AuthorDate: Thu Oct 22 16:09:28 2026 +0200
CAMEL-25917: camel-undertow - a WebSocket connection accepted while only a
producer served the path is not treated as having passed a later consumer's
custom handlers (#28237)
* CAMEL-25917: camel-undertow - a WebSocket connection accepted while only
a producer served the path is not treated as having passed a later consumer's
custom handlers
* CAMEL-25917: close WebSocket peers that no longer pass the path checks
when the consumer starts
Close the already-open connections of a path right away when a consumer is
set, rather than only when they next send a message, so a receive-only
client
is told to reconnect instead of silently receiving nothing after the
consumer
with custom handlers starts. Add a regression test that awaits the close
without the client sending, and adjust the existing window tests for the now
proactive close.
* CAMEL-25917: camel-undertow - make the window tests tolerate the
proactive close
The consumer now closes a connection opened before it started as soon as it
starts, so sending on that connection afterwards is best effort and bounded.
---
.../undertow/handlers/CamelWebSocketHandler.java | 38 ++++-
...roviderWebSocketConsumerHandlersWindowTest.java | 170 +++++++++++++++++++++
.../UndertowWsOAuthProfileConsumerWindowTest.java | 4 +-
.../ws/UndertowWsSecurityWithoutProviderTest.java | 12 +-
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 13 ++
5 files changed, 228 insertions(+), 9 deletions(-)
diff --git
a/components/camel-undertow/src/main/java/org/apache/camel/component/undertow/handlers/CamelWebSocketHandler.java
b/components/camel-undertow/src/main/java/org/apache/camel/component/undertow/handlers/CamelWebSocketHandler.java
index 9d855f8e0742..b16eb2a75314 100644
---
a/components/camel-undertow/src/main/java/org/apache/camel/component/undertow/handlers/CamelWebSocketHandler.java
+++
b/components/camel-undertow/src/main/java/org/apache/camel/component/undertow/handlers/CamelWebSocketHandler.java
@@ -245,7 +245,7 @@ public class CamelWebSocketHandler implements HttpHandler {
}
if (!authenticatedEndpoints.isEmpty()) {
exchange.putAttachment(HANDSHAKE_RESULT_ATTACHMENT,
- new HandshakeResult(authenticatedEndpoints, headers));
+ new HandshakeResult(authenticatedEndpoints, headers,
path.consumer));
}
this.delegate.handleRequest(exchange);
}
@@ -303,9 +303,14 @@ public class CamelWebSocketHandler implements HttpHandler {
return false;
}
if (requiresHandshakeResult(endpoint, consumer)) {
- return channel != null
- && channel.getAttribute(HANDSHAKE_RESULT) instanceof
HandshakeResult result
- && result.endpointUris.contains(endpoint.getEndpointUri());
+ if (channel == null
+ || !(channel.getAttribute(HANDSHAKE_RESULT) instanceof
HandshakeResult result)
+ ||
!result.endpointUris.contains(endpoint.getEndpointUri())) {
+ return false;
+ }
+ // the custom handlers run only during a consumer-mode handshake,
so a result recorded while only
+ // producers served the path does not account for them
+ return !consumer || endpoint.getHandlers() == null ||
result.consumer;
}
return true;
}
@@ -403,12 +408,30 @@ public class CamelWebSocketHandler implements HttpHandler
{
= new ConsumerHandlers(
consumer.getEndpoint(), consumerHandler !=
null ? consumerHandler : upgradeHandler);
producerEndpoints.forEach(this::warnIfProducerSettingsUnused);
+ closePeersThatNoLongerPassChecks();
}
} finally {
consumerLock.unlock();
}
}
+ /**
+ * Closes the already-open peers whose handshake did not pass the security
checks that now apply to the path. A
+ * connection accepted while only a producer served the path ran the
producer checks but not the custom handlers a
+ * consumer configures, so once that consumer is set such a peer no longer
passes the path checks. It is closed
+ * right away rather than only when it next sends a message, so that a
receive-only client is told to reconnect
+ * instead of silently receiving nothing. {@link #send} filters the same
peers out of the meantime.
+ */
+ private void closePeersThatNoLongerPassChecks() {
+ PathSecurity path = pathSecurity();
+ delegate.getPeerConnections().stream()
+ .filter(peer -> !isAuthenticated(peer, path))
+ .forEach(peer -> {
+ LOG.warn("Closing WebSocket channel whose handshake did
not pass the security checks of the consumer");
+ WebSockets.sendClose(CloseMessage.MSG_VIOLATES_POLICY,
"Authentication required", peer, null);
+ });
+ }
+
/**
* Registers the endpoint of a producer that sends to the peers of this
handler. A path without a consumer applies
* the security settings of its producers.
@@ -663,15 +686,18 @@ public class CamelWebSocketHandler implements HttpHandler
{
/**
* The endpoints whose security provider, allowed roles and custom
handlers a handshake passed, and the headers that
- * the providers added.
+ * the providers added. {@code consumer} records whether the handshake ran
in consumer mode: only then were the
+ * consumer-only custom handlers part of the checks.
*/
private static final class HandshakeResult {
private final Set<String> endpointUris;
private final Map<String, Object> headers;
+ private final boolean consumer;
- private HandshakeResult(Set<String> endpointUris, Map<String, Object>
headers) {
+ private HandshakeResult(Set<String> endpointUris, Map<String, Object>
headers, boolean consumer) {
this.endpointUris = Collections.unmodifiableSet(endpointUris);
this.headers = Collections.unmodifiableMap(headers);
+ this.consumer = consumer;
}
}
diff --git
a/components/camel-undertow/src/test/java/org/apache/camel/component/undertow/spi/SecurityProviderWebSocketConsumerHandlersWindowTest.java
b/components/camel-undertow/src/test/java/org/apache/camel/component/undertow/spi/SecurityProviderWebSocketConsumerHandlersWindowTest.java
new file mode 100644
index 000000000000..232c0c51c5f1
--- /dev/null
+++
b/components/camel-undertow/src/test/java/org/apache/camel/component/undertow/spi/SecurityProviderWebSocketConsumerHandlersWindowTest.java
@@ -0,0 +1,170 @@
+/*
+ * 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.undertow.spi;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.WebSocket;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.undertow.websockets.core.CloseMessage;
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.component.undertow.UndertowBasicAuthHandler;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A WebSocket path whose producer applies the allowed roles while its
consumer, started later, adds its own handlers on
+ * the same endpoint URI: a connection accepted while only the producer served
the path runs the allowed-roles check but
+ * not the consumer handlers, so once that consumer is up the connection is
neither served by the route nor sent
+ * messages, while a connection that goes through the consumer handlers is
served.
+ */
+class SecurityProviderWebSocketConsumerHandlersWindowTest extends
AbstractSecurityProviderTest {
+
+ private final AtomicInteger routeInvocations = new AtomicInteger();
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ context.getRegistry().bind("wsHandlers", new
UndertowBasicAuthHandler());
+ return context;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // producer and consumer share the same endpoint URI: the
producer keeps the shared handler
+ // registered and applies allowedRoles, the consumer adds the
handlers on top
+ from("direct:feed")
+
.to("undertow:ws://localhost:{{port}}/guarded?allowedRoles=user&handlers=#wsHandlers&sendToAll=true");
+
from("undertow:ws://localhost:{{port}}/guarded?allowedRoles=user&handlers=#wsHandlers&sendToAll=true")
+ .routeId("guarded").autoStartup(false)
+ .process(exchange ->
routeInvocations.incrementAndGet())
+ .to("mock:guarded");
+ }
+ };
+ }
+
+ @Test
+ void connectionAcceptedBeforeAConsumerWithHandlersStartedIsNotServed()
throws Exception {
+ securityConfiguration.setRoleToAssign("user");
+
+ // only the producer serves the path, so the handshake runs the
allowed-roles check but not the consumer handlers
+ RecordingListener listener = new RecordingListener();
+ WebSocket webSocket = connect("/guarded", null, listener);
+
+ context.getRouteController().startRoute("guarded");
+
+ // the connection does not get what the producer sends
+ template.sendBody("direct:feed", "broadcast");
+ // a message from the stale connection never reaches the route; the
send may already fail because the
+ // connection was closed when the consumer started, so it is best
effort here
+ webSocket.sendText("hello", true).whenComplete((result, error) -> {
+ });
+
+ assertEquals(CloseMessage.MSG_VIOLATES_POLICY,
listener.closeCode.orTimeout(10, TimeUnit.SECONDS).join());
+ assertTrue(listener.received.isEmpty());
+ assertEquals(0, routeInvocations.get());
+ }
+
+ @Test
+ void receiveOnlyConnectionBeforeConsumerStartsIsClosedWhenConsumerStarts()
throws Exception {
+ securityConfiguration.setRoleToAssign("user");
+
+ // only the producer serves the path: the connection passes the
allowed-roles check but not the consumer handlers
+ RecordingListener listener = new RecordingListener();
+ connect("/guarded", null, listener);
+
+ // a first broadcast confirms the peer is registered server-side while
only the producer serves the path
+ template.sendBody("direct:feed", "before");
+ assertEquals("before", listener.firstText.orTimeout(10,
TimeUnit.SECONDS).join());
+
+ context.getRouteController().startRoute("guarded");
+
+ // the client sends nothing: a receive-only connection must still be
closed when the consumer starts, so that
+ // reconnect-on-close logic runs instead of the client silently
receiving no further broadcasts
+ assertEquals(CloseMessage.MSG_VIOLATES_POLICY,
listener.closeCode.orTimeout(10, TimeUnit.SECONDS).join());
+
+ // a broadcast after the close does not reach the stale connection
+ template.sendBody("direct:feed", "after");
+ assertEquals(List.of("before"), listener.received);
+ assertEquals(0, routeInvocations.get());
+ }
+
+ @Test
+ void connectionThatRunsTheConsumerHandlersIsServed() throws Exception {
+ securityConfiguration.setRoleToAssign("user");
+ context.getRouteController().startRoute("guarded");
+
+ getMockEndpoint("mock:guarded").expectedBodiesReceived("hello");
+ String credentials =
Base64.getEncoder().encodeToString("guest:secret".getBytes(StandardCharsets.UTF_8));
+ WebSocket webSocket = connect("/guarded", "Basic " + credentials, new
RecordingListener());
+ webSocket.sendText("hello", true).join();
+
+ MockEndpoint.assertIsSatisfied(context);
+ assertEquals(1, routeInvocations.get());
+ webSocket.sendClose(WebSocket.NORMAL_CLOSURE, "done").join();
+ }
+
+ private WebSocket connect(String path, String authorization,
WebSocket.Listener listener) {
+ WebSocket.Builder builder =
HttpClient.newHttpClient().newWebSocketBuilder();
+ if (authorization != null) {
+ builder.header("Authorization", authorization);
+ }
+ return builder.buildAsync(URI.create("ws://localhost:" + getPort() +
path), listener)
+ .orTimeout(5, TimeUnit.SECONDS).join();
+ }
+
+ private static final class RecordingListener implements WebSocket.Listener
{
+
+ private final List<String> received = new CopyOnWriteArrayList<>();
+ private final CompletableFuture<String> firstText = new
CompletableFuture<>();
+ private final CompletableFuture<Integer> closeCode = new
CompletableFuture<>();
+
+ @Override
+ public void onOpen(WebSocket webSocket) {
+ webSocket.request(1);
+ }
+
+ @Override
+ public CompletionStage<?> onText(WebSocket webSocket, CharSequence
data, boolean last) {
+ received.add(data.toString());
+ firstText.complete(data.toString());
+ webSocket.request(1);
+ return null;
+ }
+
+ @Override
+ public CompletionStage<?> onClose(WebSocket webSocket, int statusCode,
String reason) {
+ closeCode.complete(statusCode);
+ return null;
+ }
+ }
+}
diff --git
a/components/camel-undertow/src/test/java/org/apache/camel/component/undertow/ws/UndertowWsOAuthProfileConsumerWindowTest.java
b/components/camel-undertow/src/test/java/org/apache/camel/component/undertow/ws/UndertowWsOAuthProfileConsumerWindowTest.java
index 06a19be0e1bc..d71fa3a886ca 100644
---
a/components/camel-undertow/src/test/java/org/apache/camel/component/undertow/ws/UndertowWsOAuthProfileConsumerWindowTest.java
+++
b/components/camel-undertow/src/test/java/org/apache/camel/component/undertow/ws/UndertowWsOAuthProfileConsumerWindowTest.java
@@ -69,7 +69,9 @@ public class UndertowWsOAuthProfileConsumerWindowTest extends
BaseUndertowTest {
context.getRouteController().startRoute("secureWs");
template.sendBody("direct:keepalive", "broadcast");
- webSocket.sendText("hello", true).orTimeout(5,
TimeUnit.SECONDS).join();
+ // the channel is closed when the secured consumer starts, so a send
from it may already fail: best effort
+ webSocket.sendText("hello", true).whenComplete((result, error) -> {
+ });
assertEquals(CloseMessage.MSG_VIOLATES_POLICY,
listener.closeCode.orTimeout(10, TimeUnit.SECONDS).join());
assertTrue(listener.received.isEmpty());
diff --git
a/components/camel-undertow/src/test/java/org/apache/camel/component/undertow/ws/UndertowWsSecurityWithoutProviderTest.java
b/components/camel-undertow/src/test/java/org/apache/camel/component/undertow/ws/UndertowWsSecurityWithoutProviderTest.java
index 09f301add9ea..ce15e4e87b60 100644
---
a/components/camel-undertow/src/test/java/org/apache/camel/component/undertow/ws/UndertowWsSecurityWithoutProviderTest.java
+++
b/components/camel-undertow/src/test/java/org/apache/camel/component/undertow/ws/UndertowWsSecurityWithoutProviderTest.java
@@ -128,7 +128,11 @@ class UndertowWsSecurityWithoutProviderTest extends
BaseUndertowTest {
// the connection does not get what the producer sends
template.sendBody("direct:late", "broadcast");
// and its messages do not reach the route: it is closed instead
- webSocket.sendText("hello", true).join();
+ // The consumer may already have closed this stale connection; sending
is best effort.
+ webSocket.sendText("hello", true)
+ .orTimeout(5, TimeUnit.SECONDS)
+ .handle((result, error) -> null)
+ .join();
assertEquals(CloseMessage.MSG_VIOLATES_POLICY,
listener.closeCode.orTimeout(10, TimeUnit.SECONDS).join());
assertTrue(listener.received.isEmpty());
@@ -147,7 +151,11 @@ class UndertowWsSecurityWithoutProviderTest extends
BaseUndertowTest {
// the connection does not get what the producer sends
template.sendBody("direct:lateBasic", "broadcast");
// and its messages do not reach the route: it is closed instead
- webSocket.sendText("hello", true).join();
+ // The consumer may already have closed this stale connection; sending
is best effort.
+ webSocket.sendText("hello", true)
+ .orTimeout(5, TimeUnit.SECONDS)
+ .handle((result, error) -> null)
+ .join();
assertEquals(CloseMessage.MSG_VIOLATES_POLICY,
listener.closeCode.orTimeout(10, TimeUnit.SECONDS).join());
assertTrue(listener.received.isEmpty());
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 46379cf15944..29ad36b1652f 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -1515,6 +1515,19 @@ framing, are still allowed. Routes that put a
percent-encoded control character
`CamelHttpPath` or `CamelHttpQuery` header, which the producer decodes before
building the request, are the ones most
likely to be affected.
+=== camel-undertow - a WebSocket connection opened before a consumer with
custom handlers started is no longer served
+
+On a WebSocket path whose producer applies a security provider or allowed
roles, a connection could be accepted while
+only the producer served the path, before its consumer was started for the
first time. Such a connection ran the
+producer checks but not the custom `handlers` the consumer configures, as
those handlers run only for consumers. Once
+the consumer was started, the connection was still treated as having passed
the consumer checks, so its messages
+reached the route and it kept receiving what the producer sent.
+
+Once a consumer that configures `handlers` is started on a path, a connection
opened on that path before the consumer
+first ran is no longer served: its messages no longer reach the route and it
no longer receives what the producer
+sends, and the connection is closed. Clients that open a connection in this
window must reconnect after the consumer
+has started.
+
=== camel-dapr - the saveBulk and executeTransaction state operations are sent
The `saveBulk` and `executeTransaction` operations of `dapr:state` built their
request but never sent it to Dapr, so
