Ignore invalid STOMP frame

Closes spring-projectsgh-28444

Author: rstoyanchev

spring-messaging/src/main/java/org/springframework/messaging/simp/broker/SimpleBrokerMessageHandler.java

@@ -296,18 +296,29 @@ protected void handleMessageInternal(Message<?> message) {
 		}
 		else if (SimpMessageType.CONNECT.equals(messageType)) {
 			logMessage(message);
-			long[] clientHeartbeat = SimpMessageHeaderAccessor.getHeartbeat(headers);
-			long[] serverHeartbeat = getHeartbeatValue();
-			Principal user = SimpMessageHeaderAccessor.getUser(headers);
-			this.sessions.put(sessionId, new SessionInfo(sessionId, user, clientHeartbeat, serverHeartbeat));
-			SimpMessageHeaderAccessor connectAck = SimpMessageHeaderAccessor.create(SimpMessageType.CONNECT_ACK);
-			initHeaders(connectAck);
-			connectAck.setSessionId(sessionId);
-			connectAck.setUser(SimpMessageHeaderAccessor.getUser(headers));
-			connectAck.setHeader(SimpMessageHeaderAccessor.CONNECT_MESSAGE_HEADER, message);
-			connectAck.setHeader(SimpMessageHeaderAccessor.HEART_BEAT_HEADER, serverHeartbeat);
-			Message<byte[]> messageOut = MessageBuilder.createMessage(EMPTY_PAYLOAD, connectAck.getMessageHeaders());
-			getClientOutboundChannel().send(messageOut);
+
+			if (sessionId != null) {
+				if (this.sessions.get(sessionId) != null) {
+					if (logger.isWarnEnabled()) {
+						logger.warn("Ignoring CONNECT in session " + sessionId + ". Already connected.");
+					}
+					return;
+				}
+
+				long[] clientHeartbeat = SimpMessageHeaderAccessor.getHeartbeat(headers);
+				long[] serverHeartbeat = getHeartbeatValue();
+				Principal user = SimpMessageHeaderAccessor.getUser(headers);
+				this.sessions.put(sessionId, new SessionInfo(sessionId, user, clientHeartbeat, serverHeartbeat));
+				SimpMessageHeaderAccessor connectAck = SimpMessageHeaderAccessor.create(SimpMessageType.CONNECT_ACK);
+				initHeaders(connectAck);
+				connectAck.setSessionId(sessionId);
+				connectAck.setUser(SimpMessageHeaderAccessor.getUser(headers));
+				connectAck.setHeader(SimpMessageHeaderAccessor.CONNECT_MESSAGE_HEADER, message);
+				connectAck.setHeader(SimpMessageHeaderAccessor.HEART_BEAT_HEADER, serverHeartbeat);
+				Message<byte[]> messageOut =
+						MessageBuilder.createMessage(EMPTY_PAYLOAD, connectAck.getMessageHeaders());
+				getClientOutboundChannel().send(messageOut);
+			}
 		}
 		else if (SimpMessageType.DISCONNECT.equals(messageType)) {
 			logMessage(message);

spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java

@@ -493,7 +493,13 @@ else if (accessor instanceof SimpMessageHeaderAccessor) {
 			return;
 		}
 
-		if (StompCommand.CONNECT.equals(command)) {
+		if (StompCommand.CONNECT.equals(command) || StompCommand.STOMP.equals(command)) {
+			if (this.connectionHandlers.get(sessionId) != null) {
+				if (logger.isWarnEnabled()) {
+					logger.warn("Ignoring CONNECT in session " + sessionId + ". Already connected.");
+				}
+				return;
+			}
 			if (logger.isDebugEnabled()) {
 				logger.debug(stompAccessor.getShortLogMessage(EMPTY_PAYLOAD));
 			}

spring-messaging/src/test/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandlerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2015 the original author or authors.
+ * Copyright 2002-2019 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -220,6 +220,34 @@ public void systemSubscription() throws Exception {
 		assertSame(message, captor.getValue());
 	}
 
+
+
+	@Test
+	public void alreadyConnected() throws Exception{
+
+		this.brokerRelay.start();
+
+		Message<byte[]> connect = connectMessage("sess1", "joe");
+		this.brokerRelay.handleMessage(connect);
+
+		assertEquals(2, this.tcpClient.getSentMessages().size());
+
+		StompHeaderAccessor headers1 = this.tcpClient.getSentHeaders(0);
+		assertEquals(StompCommand.CONNECT, headers1.getCommand());
+		assertEquals(StompBrokerRelayMessageHandler.SYSTEM_SESSION_ID, headers1.getSessionId());
+
+
+		StompHeaderAccessor headers2 = this.tcpClient.getSentHeaders(1);
+		assertEquals(StompCommand.CONNECT, headers2.getCommand());
+		assertEquals("sess1", headers2.getSessionId());
+
+		this.brokerRelay.handleMessage(connect);
+
+		assertEquals(2, this.tcpClient.getSentMessages().size());
+		assertTrue(this.outboundChannel.getMessages().isEmpty());
+	}
+
+
 	private Message<byte[]> connectMessage(String sessionId, String user) {
 		StompHeaderAccessor headers = StompHeaderAccessor.create(StompCommand.CONNECT);
 		headers.setSessionId(sessionId);