From a8f0b56805c45de05a2919cefefa19840d38aaf5 Mon Sep 17 00:00:00 2001 From: Federico Mariani <34543311+Croway@users.noreply.github.com> Date: Thu, 22 Oct 2026 16:09:28 +0200 Subject: [PATCH] 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) --- .../handlers/CamelWebSocketHandler.java | 38 +++- ...erWebSocketConsumerHandlersWindowTest.java | 170 ++++++++++++++++++ ...UndertowWsSecurityWithoutProviderTest.java | 12 +- 3 files changed, 212 insertions(+), 8 deletions(-) create mode 100644 components/camel-undertow/src/test/java/org/apache/camel/component/undertow/spi/SecurityProviderWebSocketConsumerHandlersWindowTest.java 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 39e8b8e417393..2bae0c9c55c30 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 @@ private void upgrade(HttpServerExchange exchange) throws Exception { } 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 static boolean isAuthenticated(WebSocketChannel channel, UndertowEndpoint 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 void setConsumer(UndertowConsumer consumer, HttpHandler consumerHandler) = 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 @@ private ConsumerHandlers(UndertowEndpoint endpoint, HttpHandler handler) { /** * 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 endpointUris; private final Map headers; + private final boolean consumer; - private HandshakeResult(Set endpointUris, Map headers) { + private HandshakeResult(Set endpointUris, Map 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 0000000000000..232c0c51c5f10 --- /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 received = new CopyOnWriteArrayList<>(); + private final CompletableFuture firstText = new CompletableFuture<>(); + private final CompletableFuture 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 9f7ee35ec2593..0a4fb198bee8c 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 @@ void connectionOpenedBeforeTheConsumerStartedIsNotServed() throws Exception { // 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 @@ void connectionOpenedBeforeAConsumerWithHandlersStartedIsNotServed() throws Exce // 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());