This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch camel-4.18.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.18.x by this push:
new 03ff3660c72b [camel-4.18.x] 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 (#28506)
03ff3660c72b is described below
commit 03ff3660c72b7f4372c71509a4e0a80c8a0bbb6a
Author: Federico Mariani <[email protected]>
AuthorDate: Mon Oct 26 11:39:01 2026 +0100
[camel-4.18.x] 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 (#28506)
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.
(cherry picked from commit 56aa5f4970a6fb7126275fc02b5bb9694dd184d1)
---
.../undertow/handlers/CamelWebSocketHandler.java | 38 ++++-
...roviderWebSocketConsumerHandlersWindowTest.java | 170 +++++++++++++++++++++
.../ws/UndertowWsSecurityWithoutProviderTest.java | 12 +-
3 files changed, 212 insertions(+), 8 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 39e8b8e41739..2bae0c9c55c3 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
@@ -225,7 +225,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);
}
@@ -274,9 +274,14 @@ public class CamelWebSocketHandler implements HttpHandler {
private static boolean isAuthenticated(WebSocketChannel channel,
UndertowEndpoint endpoint, boolean consumer) {
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;
}
@@ -353,12 +358,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.
@@ -613,15 +636,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/UndertowWsSecurityWithoutProviderTest.java
b/components/camel-undertow/src/test/java/org/apache/camel/component/undertow/ws/UndertowWsSecurityWithoutProviderTest.java
index 9f7ee35ec259..0a4fb198bee8 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
@@ -117,7 +117,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());
@@ -136,7 +140,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());
