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

components/camel-undertow/src/main/java/org/apache/camel/component/undertow/handlers/CamelWebSocketHandler.java

@@ -245,7 +245,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);
     }
@@ -303,9 +303,14 @@ private static boolean isAuthenticated(WebSocketChannel channel, UndertowEndpoin
             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 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.
@@ -663,15 +686,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<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;
         }
     }
 

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;
+        }
+    }
+}

components/camel-undertow/src/test/java/org/apache/camel/component/undertow/ws/UndertowWsOAuthProfileConsumerWindowTest.java

@@ -69,7 +69,9 @@ public void closesChannelEstablishedBeforeSecuredConsumerStarted() throws Except
         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());

components/camel-undertow/src/test/java/org/apache/camel/component/undertow/ws/UndertowWsSecurityWithoutProviderTest.java

@@ -128,7 +128,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());
@@ -147,7 +151,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());

docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc

@@ -1376,6 +1376,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