CVE-2023-32081: Improper Authentication (CWE-287)
value: Vert.x STOMP is a vert.x implementation of the STOMP specification that provides a STOMP server and client. From versions 3.1.0 until 3.9.16 and 4.0.0 until 4.4.2, a Vert.x STOMP server processes client STOMP frames without checking that the client send an initial CONNECT frame replied with a successful CONNECTED frame. The client can subscribe to a destination or publish message without prior authentication. Any Vert.x STOMP server configured with an authentication handler is impacted. The issue is patched in Vert.x 3.9.16 and 4.4.2. There are no trivial workarounds.
CWE description: When an actor claims to have a given identity, the product does not prove or insufficiently proves that the claim is correct.
CVSS v3 base score: 6.5
Published: 2023-05-12T14:15Z
Repository: https://github.com/vert-x3/vertx-stomp
Language: Java
Files changed: src/main/java/io/vertx/ext/stomp/DefaultConnectHandler.java, src/main/java/io/vertx/ext/stomp/StompServerConnection.java, src/main/java/io/vertx/ext/stomp/impl/DefaultStompHandler.java, src/main/java/io/vertx/ext/stomp/impl/StompServerImpl.java, src/test/java/io/vertx/ext/stomp/impl/SecuredServerConnectionTest.java
Commit date: 2023-05-12 10:40:36 +0200
Commit message: STOMP server process client frames that would not send initially a connect frame

A Vert.x STOMP server processes client STOMP frames without checking that the client send an initial CONNECT frame replied with a successful CONNECTED frame. The client can subscribe to a destination or publish message without prior authentication. Any Vert.x STOMP server configured with an authentication handler is impacted.

Fixes CVE-2023-32081

Diff: diff --git a/src/main/java/io/vertx/ext/stomp/DefaultConnectHandler.java b/src/main/java/io/vertx/ext/stomp/DefaultConnectHandler.java
index f040867..fa2344d 100644
--- a/src/main/java/io/vertx/ext/stomp/DefaultConnectHandler.java
+++ b/src/main/java/io/vertx/ext/stomp/DefaultConnectHandler.java
@@ -37,10 +37,11 @@ import java.util.List;
  * This handler is thread safe.
  *
  * @author <a href="http://escoffier.me">Clement Escoffier</a>
  */
 public class DefaultConnectHandler implements Handler<ServerFrame> {
+
   @Override
   public void handle(ServerFrame sf) {
     // Server negotiation
     List<String> accepted = new ArrayList<>();
     String accept = sf.frame().getHeader(Frame.ACCEPT_VERSION);
diff --git a/src/main/java/io/vertx/ext/stomp/StompServerConnection.java b/src/main/java/io/vertx/ext/stomp/StompServerConnection.java
index 3aadcab..db25eef 100644
--- a/src/main/java/io/vertx/ext/stomp/StompServerConnection.java
+++ b/src/main/java/io/vertx/ext/stomp/StompServerConnection.java
@@ -94,6 +94,7 @@ public interface StompServerConnection {
    * @param ping ping time
    * @param pong pong time
    * @param pingHandler the ping handler
    */
   void configureHeartbeat(long ping, long pong, Handler<StompServerConnection> pingHandler);
+
 }
diff --git a/src/main/java/io/vertx/ext/stomp/impl/DefaultStompHandler.java b/src/main/java/io/vertx/ext/stomp/impl/DefaultStompHandler.java
index 7f11ad4..e0c10bc 100644
--- a/src/main/java/io/vertx/ext/stomp/impl/DefaultStompHandler.java
+++ b/src/main/java/io/vertx/ext/stomp/impl/DefaultStompHandler.java
@@ -68,11 +68,11 @@ public class DefaultStompHandler implements StompServerHandler {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(DefaultStompHandler.class);
   private final Vertx vertx;
   private final Context context;
 
-  private Handler<ServerFrame> connectHandler = new DefaultConnectHandler();
+  private Handler<ServerFrame> connectHandler;
 
   private Handler<ServerFrame> stompHandler;
 
   private Handler<ServerFrame> sendHandler = new DefaultSendHandler();
 
@@ -123,10 +123,11 @@ public class DefaultStompHandler implements StompServerHandler {
   public DefaultStompHandler(Vertx vertx) {
     this.vertx = vertx;
     this.context = Vertx.currentContext();
     this.destinations = vertx.sharedData().getLocalMap("stomp.destinations");
     this.users = new ConcurrentHashMap<>();
+    this.connectHandler = new DefaultConnectHandler();
   }
 
   @Override
   public synchronized void onClose(StompServerConnection connection) {
     // Default behavior.
diff --git a/src/main/java/io/vertx/ext/stomp/impl/StompServerImpl.java b/src/main/java/io/vertx/ext/stomp/impl/StompServerImpl.java
index fa722ae..2071507 100644
--- a/src/main/java/io/vertx/ext/stomp/impl/StompServerImpl.java
+++ b/src/main/java/io/vertx/ext/stomp/impl/StompServerImpl.java
@@ -25,11 +25,13 @@ import io.vertx.core.http.ServerWebSocket;
 import io.vertx.core.impl.logging.Logger;
 import io.vertx.core.impl.logging.LoggerFactory;
 import io.vertx.core.net.NetServer;
 import io.vertx.ext.stomp.*;
 
+import java.util.Collections;
 import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * Default implementation of the {@link StompServer}.
  *
  * @author <a href="http://escoffier.me">Clement Escoffier</a>
@@ -107,11 +109,21 @@ public class StompServerImpl implements StompServer {
 
     Objects.requireNonNull(stomp, "Cannot open STOMP server - no StompServerConnectionHandler attached to the " +
         "server.");
     server
         .connectHandler(socket -> {
-          StompServerConnection connection = new StompServerTCPConnectionImpl(socket, this, writingFrameHandler);
+          AtomicBoolean connected = new AtomicBoolean();
+          AtomicBoolean firstFrame = new AtomicBoolean();
+          StompServerConnection connection = new StompServerTCPConnectionImpl(socket, this, frame -> {
+            if (frame.frame().getCommand() == Command.CONNECTED) {
+              connected.set(true);
+            }
+            Handler<ServerFrame> h = writingFrameHandler;
+            if (h != null) {
+              h.handle(frame);
+            }
+          });
           FrameParser parser = new FrameParser(options);
           socket.exceptionHandler((exception) -> {
             LOGGER.error("The STOMP server caught a TCP socket error - closing connection", exception);
             connection.close();
           });
@@ -121,11 +133,25 @@ public class StompServerImpl implements StompServer {
                     connection.write(
                         Frames.createInvalidFrameErrorFrame(exception));
                     connection.close();
                   }
               )
-              .handler(frame -> stomp.handle(new ServerFrameImpl(frame, connection)));
+              .handler(frame -> {
+                if (frame.getCommand() == Command.CONNECT || frame.getCommand() == Command.STOMP) {
+                  if (firstFrame.compareAndSet(false, true)) {
+                    stomp.handle(new ServerFrameImpl(frame, connection));
+                  } else {
+                    connection.write(Frames.createErrorFrame("Already connected", Collections.emptyMap(), ""));
+                    connection.close();
+                  }
+                } else if (connected.get()) {
+                  stomp.handle(new ServerFrameImpl(frame, connection));
+                } else {
+                  connection.write(Frames.createErrorFrame("Not connected", Collections.emptyMap(), ""));
+                  connection.close();
+                }
+              });
           socket.handler(parser);
         })
         .listen(port, host).onComplete(ar -> {
           if (ar.failed()) {
             if (handler != null) {
@@ -216,11 +242,21 @@ public class StompServerImpl implements StompServer {
         LOGGER.error("Receiving a web socket connection on an invalid path (" + socket.path() + "), the path is " +
             "configured to " + options.getWebsocketPath() + ". Rejecting connection");
         socket.reject();
         return;
       }
-      StompServerConnection connection = new StompServerWebSocketConnectionImpl(socket, this, writingFrameHandler);
+      AtomicBoolean connected = new AtomicBoolean();
+      AtomicBoolean firstFrame = new AtomicBoolean();
+      StompServerConnection connection = new StompServerWebSocketConnectionImpl(socket, this, frame -> {
+        if (frame.frame().getCommand() == Command.CONNECTED  || frame.frame().getCommand() == Command.STOMP) {
+          connected.set(true);
+        }
+        Handler<ServerFrame> h = writingFrameHandler;
+        if (h != null) {
+          h.handle(frame);
+        }
+      });
       FrameParser parser = new FrameParser(options);
       socket.exceptionHandler((exception) -> {
         LOGGER.error("The STOMP server caught a WebSocket error - closing connection", exception);
         connection.close();
       });
@@ -230,11 +266,25 @@ public class StompServerImpl implements StompServer {
                 connection.write(
                     Frames.createInvalidFrameErrorFrame(exception));
                 connection.close();
               }
           )
-          .handler(frame -> stomp.handle(new ServerFrameImpl(frame, connection)));
+          .handler(frame -> {
+            if (frame.getCommand() == Command.CONNECT) {
+              if (firstFrame.compareAndSet(false, true)) {
+                stomp.handle(new ServerFrameImpl(frame, connection));
+              } else {
+                connection.write(Frames.createErrorFrame("Already connected", Collections.emptyMap(), ""));
+                connection.close();
+              }
+            } else if (connected.get()) {
+              stomp.handle(new ServerFrameImpl(frame, connection));
+            } else {
+              connection.write(Frames.createErrorFrame("Not connected", Collections.emptyMap(), ""));
+              connection.close();
+            }
+          });
       socket.handler(parser);
     };
   }
 
   @Override
diff --git a/src/test/java/io/vertx/ext/stomp/impl/SecuredServerConnectionTest.java b/src/test/java/io/vertx/ext/stomp/impl/SecuredServerConnectionTest.java
index 9fe57d8..e30aa1b 100644
--- a/src/test/java/io/vertx/ext/stomp/impl/SecuredServerConnectionTest.java
+++ b/src/test/java/io/vertx/ext/stomp/impl/SecuredServerConnectionTest.java
@@ -14,12 +14,18 @@
  *  You may elect to redistribute this code under either of these licenses.
  */
 
 package io.vertx.ext.stomp.impl;
 
+import io.vertx.core.Future;
+import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
 import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.HttpClient;
+import io.vertx.core.http.HttpServer;
+import io.vertx.core.http.HttpServerOptions;
+import io.vertx.core.net.NetClient;
 import io.vertx.core.net.NetSocket;
 import io.vertx.ext.auth.User;
 import io.vertx.ext.auth.authentication.AuthenticationProvider;
 import io.vertx.ext.auth.properties.PropertyFileAuthentication;
 import io.vertx.ext.stomp.Frame;
@@ -36,30 +42,44 @@ import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
+import java.util.Arrays;
+
 /**
  * Tests STOMP server with security.
  *
  * @author <a href="http://escoffier.me">Clement Escoffier</a>
  */
 @RunWith(VertxUnitRunner.class)
 public class SecuredServerConnectionTest {
+
   private Vertx vertx;
   private StompServer server;
+  private HttpServer wsServer;
+  private HttpClient wsClient;
+  private StompClient client;
 
   @Rule
   public RunTestOnContext rule = new RunTestOnContext();
 
   @Before
   public void setUp(TestContext context) {
     vertx = rule.vertx();
     AuthenticationProvider provider = PropertyFileAuthentication.create(vertx, "test-auth.properties");
-    server = StompServer.create(vertx, new StompServerOptions().setSecured(true))
-        .handler(StompServerHandler.create(vertx).authProvider(provider));
-    server.listen().onComplete(context.asyncAssertSuccess());
+    server = StompServer.create(vertx, new StompServerOptions()
+        .setSecured(true)
+        .setWebsocketBridge(true)
+        .setWebsocketPath("/stomp"))
+      .handler(StompServerHandler.create(vertx).authProvider(provider));
+    server.listen(StompServerOptions.DEFAULT_STOMP_PORT).onComplete(context.asyncAssertSuccess());
+    wsServer = vertx.createHttpServer(new HttpServerOptions().setWebSocketSubProtocols(Arrays.asList("v10.stomp", "v11.stomp")))
+      .webSocketHandler(server.webSocketHandler());
+    wsServer.listen(8080).onComplete(context.asyncAssertSuccess());
+    wsClient = vertx.createHttpClient();
+    client = StompClient.create(vertx, new StompClientOptions().setLogin("admin").setPasscode("admin"));
   }
 
   @After
   public void tearDown(TestContext context) {
     server.close().onComplete(context.asyncAssertSuccess());
@@ -160,13 +180,68 @@ public class SecuredServerConnectionTest {
     StompClient.create(vertx).connect(server.actualPort(), "localhost").onComplete(context.asyncAssertFailure(err -> {
     }));
   }
 
   void validate(TestContext context, Buffer buffer) {
-    context.assertTrue(buffer.toString().contains("CONNECTED"));
+    context.assertTrue(buffer.toString().contains("CONNECTED"), "Was expected <" + buffer.toString() + "> to contain 'CONNECTED'");
     context.assertTrue(buffer.toString().contains("version:1.2"));
 
     User user = server.stompHandler().getUserBySession(extractSession(buffer.toString()));
     context.assertNotNull(user);
   }
 
+  @Test
+  public void testTCPClientMustBeConnected(TestContext context) {
+    Async async = context.async();
+    NetClient client = vertx.createNetClient();
+    testClientMustBeConnected(context, v -> {
+      client.connect(server.actualPort(), "0.0.0.0").onComplete(context.asyncAssertSuccess(so -> {
+        Buffer received = Buffer.buffer();
+        so.handler(received::appendBuffer);
+        so.write(
+          "SEND\n" +
+            "destination:/test\n" +
+            "\n" +
+            "hello" +
+            FrameParser.NULL);
+        so.endHandler(v2 -> {
+          context.assertTrue(received.toString().startsWith("ERROR\n"));
+          async.complete();
+        });
+      }));
+    });
+  }
+
+  @Test
+  public void testWebSocketClientMustBeConnected(TestContext context) {
+    Async async = context.async();
+    testClientMustBeConnected(context, v -> {
+      wsClient.webSocket(8080, "localhost", "/stomp").onComplete(context.asyncAssertSuccess(ws -> {
+        Buffer received = Buffer.buffer();
+        ws.binaryMessageHandler(received::appendBuffer);
+        ws.writeBinaryMessage(
+          Buffer.buffer("SEND\n" +
+            "destination:/test\n" +
+            "\n" +
+            "hello" +
+            FrameParser.NULL));
+        ws.endHandler(v2 -> {
+          context.assertTrue(received.toString().startsWith("ERROR\n"));
+          async.complete();
+        });
+      }));
+    });
+  }
+
+  private void testClientMustBeConnected(TestContext context, Handler<Void> cont) {
+    client
+      .connect(server.actualPort(), "localhost")
+      .onComplete(context.asyncAssertSuccess(conn -> {
+        Future<String> fut = conn.subscribe("/test", frame -> {
+          context.fail("Should not receive a messsage");
+        });
+        fut.onComplete(context.asyncAssertSuccess(v2 -> {
+          cont.handle(null);
+        }));
+      }));
+  }
 }

Vulnerable code:   private Handler<ServerFrame> connectHandler = new DefaultConnectHandler();
          StompServerConnection connection = new StompServerTCPConnectionImpl(socket, this, writingFrameHandler);
              .handler(frame -> stomp.handle(new ServerFrameImpl(frame, connection)));
      StompServerConnection connection = new StompServerWebSocketConnectionImpl(socket, this, writingFrameHandler);
          .handler(frame -> stomp.handle(new ServerFrameImpl(frame, connection)));
    server = StompServer.create(vertx, new StompServerOptions().setSecured(true))
        .handler(StompServerHandler.create(vertx).authProvider(provider));
    server.listen().onComplete(context.asyncAssertSuccess());
    context.assertTrue(buffer.toString().contains("CONNECTED"));
Fixed code: 

  private Handler<ServerFrame> connectHandler;
    this.connectHandler = new DefaultConnectHandler();
import java.util.Collections;
import java.util.concurrent.atomic.AtomicBoolean;
          AtomicBoolean connected = new AtomicBoolean();
          AtomicBoolean firstFrame = new AtomicBoolean();
          StompServerConnection connection = new StompServerTCPConnectionImpl(socket, this, frame -> {
            if (frame.frame().getCommand() == Command.CONNECTED) {
              connected.set(true);
            }
            Handler<ServerFrame> h = writingFrameHandler;
            if (h != null) {
              h.handle(frame);
            }
          });
              .handler(frame -> {
                if (frame.getCommand() == Command.CONNECT || frame.getCommand() == Command.STOMP) {
                  if (firstFrame.compareAndSet(false, true)) {
                    stomp.handle(new ServerFrameImpl(frame, connection));
                  } else {
                    connection.write(Frames.createErrorFrame("Already connected", Collections.emptyMap(), ""));
                    connection.close();
                  }
                } else if (connected.get()) {
                  stomp.handle(new ServerFrameImpl(frame, connection));
                } else {
                  connection.write(Frames.createErrorFrame("Not connected", Collections.emptyMap(), ""));
                  connection.close();
                }
              });
      AtomicBoolean connected = new AtomicBoolean();
      AtomicBoolean firstFrame = new AtomicBoolean();
      StompServerConnection connection = new StompServerWebSocketConnectionImpl(socket, this, frame -> {
        if (frame.frame().getCommand() == Command.CONNECTED  || frame.frame().getCommand() == Command.STOMP) {
          connected.set(true);
        }
        Handler<ServerFrame> h = writingFrameHandler;
        if (h != null) {
          h.handle(frame);
        }
      });
          .handler(frame -> {
            if (frame.getCommand() == Command.CONNECT) {
              if (firstFrame.compareAndSet(false, true)) {
                stomp.handle(new ServerFrameImpl(frame, connection));
              } else {
                connection.write(Frames.createErrorFrame("Already connected", Collections.emptyMap(), ""));
                connection.close();
              }
            } else if (connected.get()) {
              stomp.handle(new ServerFrameImpl(frame, connection));
            } else {
              connection.write(Frames.createErrorFrame("Not connected", Collections.emptyMap(), ""));
              connection.close();
            }
          });
import io.vertx.core.Future;
import io.vertx.core.Handler;
import io.vertx.core.http.HttpClient;
import io.vertx.core.http.HttpServer;
import io.vertx.core.http.HttpServerOptions;
import io.vertx.core.net.NetClient;
import java.util.Arrays;


  private HttpServer wsServer;
  private HttpClient wsClient;
  private StompClient client;
    server = StompServer.create(vertx, new StompServerOptions()
        .setSecured(true)
        .setWebsocketBridge(true)
        .setWebsocketPath("/stomp"))
      .handler(StompServerHandler.create(vertx).authProvider(provider));
    server.listen(StompServerOptions.DEFAULT_STOMP_PORT).onComplete(context.asyncAssertSuccess());
    wsServer = vertx.createHttpServer(new HttpServerOptions().setWebSocketSubProtocols(Arrays.asList("v10.stomp", "v11.stomp")))
      .webSocketHandler(server.webSocketHandler());
    wsServer.listen(8080).onComplete(context.asyncAssertSuccess());
    wsClient = vertx.createHttpClient();
    client = StompClient.create(vertx, new StompClientOptions().setLogin("admin").setPasscode("admin"));
    context.assertTrue(buffer.toString().contains("CONNECTED"), "Was expected <" + buffer.toString() + "> to contain 'CONNECTED'");
  @Test
  public void testTCPClientMustBeConnected(TestContext context) {
    Async async = context.async();
    NetClient client = vertx.createNetClient();
    testClientMustBeConnected(context, v -> {
      client.connect(server.actualPort(), "0.0.0.0").onComplete(context.asyncAssertSuccess(so -> {
        Buffer received = Buffer.buffer();
        so.handler(received::appendBuffer);
        so.write(
          "SEND\n" +
            "destination:/test\n" +
            "\n" +
            "hello" +
            FrameParser.NULL);
        so.endHandler(v2 -> {
          context.assertTrue(received.toString().startsWith("ERROR\n"));
          async.complete();
        });
      }));
    });
  }

  @Test
  public void testWebSocketClientMustBeConnected(TestContext context) {
    Async async = context.async();
    testClientMustBeConnected(context, v -> {
      wsClient.webSocket(8080, "localhost", "/stomp").onComplete(context.asyncAssertSuccess(ws -> {
        Buffer received = Buffer.buffer();
        ws.binaryMessageHandler(received::appendBuffer);
        ws.writeBinaryMessage(
          Buffer.buffer("SEND\n" +
            "destination:/test\n" +
            "\n" +
            "hello" +
            FrameParser.NULL));
        ws.endHandler(v2 -> {
          context.assertTrue(received.toString().startsWith("ERROR\n"));
          async.complete();
        });
      }));
    });
  }

  private void testClientMustBeConnected(TestContext context, Handler<Void> cont) {
    client
      .connect(server.actualPort(), "localhost")
      .onComplete(context.asyncAssertSuccess(conn -> {
        Future<String> fut = conn.subscribe("/test", frame -> {
          context.fail("Should not receive a messsage");
        });
        fut.onComplete(context.asyncAssertSuccess(v2 -> {
          cont.handle(null);
        }));
      }));
  }