From c5998444cf3c29ea97cea28937180bf8b99dd666 Mon Sep 17 00:00:00 2001
From: Timothy Bish <tabish121@gmail.com>
Date: Fri, 10 Jun 2016 17:11:44 -0400
Subject: [PATCH] https://issues.apache.org/jira/browse/AMQ-6319

Additional fixes and tests for disable of non-SASL client connects.
---
 .../transport/amqp/AmqpWireFormat.java        | 15 ++++++++++++---
 .../amqp/protocol/AmqpConnection.java         |  2 +-
 .../amqp/sasl/AmqpAuthenticator.java          |  2 +-
 .../amqp/protocol/AmqpWireFormatTest.java     | 19 +++++++++++++++----
 4 files changed, 29 insertions(+), 9 deletions(-)

diff --git a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpWireFormat.java b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpWireFormat.java
index 149eb7092d..ddad5e8799 100644
--- a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpWireFormat.java
+++ b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpWireFormat.java
@@ -121,21 +121,30 @@ public class AmqpWireFormat implements WireFormat {
      * Given an AMQP header validate that the AMQP magic is present and
      * if so that the version and protocol values align with what we support.
      *
+     * In the case where authentication occurs the client sends us two AMQP
+     * headers, the first being the SASL initial header which triggers the
+     * authentication process and then if that succeeds we should get a second
+     * AMQP header that does not contain the SASL protocol ID indicating the
+     * connection process should follow the normal path.  We validate that the
+     * header align with these expectations.
+     *
      * @param header
      *        the header instance received from the client.
+     * @param authenticated
+     *        has the client already authenticated already.
      *
      * @return true if the header is valid against the current WireFormat.
      */
-    public boolean isHeaderValid(AmqpHeader header) {
+    public boolean isHeaderValid(AmqpHeader header, boolean authenticated) {
         if (!header.hasValidPrefix()) {
             return false;
         }
 
-        if (!(header.getProtocolId() == 0 || header.getProtocolId() == 3)) {
+        if (!(header.getProtocolId() == 0 || header.getProtocolId() == SASL_PROTOCOL)) {
             return false;
         }
 
-        if (!isAllowNonSaslConnections() && header.getProtocolId() != SASL_PROTOCOL) {
+        if (!authenticated && !isAllowNonSaslConnections() && header.getProtocolId() != SASL_PROTOCOL) {
             return false;
         }
 
diff --git a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/protocol/AmqpConnection.java b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/protocol/AmqpConnection.java
index a0927adcb0..8b86132ffd 100644
--- a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/protocol/AmqpConnection.java
+++ b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/protocol/AmqpConnection.java
@@ -327,7 +327,7 @@ public class AmqpConnection implements AmqpProtocolConverter {
         if (command.getClass() == AmqpHeader.class) {
             AmqpHeader header = (AmqpHeader) command;
 
-            if (amqpWireFormat.isHeaderValid(header)) {
+            if (amqpWireFormat.isHeaderValid(header, authenticator != null)) {
                 LOG.trace("Connection from an AMQP v1.0 client initiated. {}", header);
             } else {
                 LOG.warn("Connection attempt from non AMQP v1.0 client. {}", header);
diff --git a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/sasl/AmqpAuthenticator.java b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/sasl/AmqpAuthenticator.java
index 5ae94888b4..379c5f995a 100644
--- a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/sasl/AmqpAuthenticator.java
+++ b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/sasl/AmqpAuthenticator.java
@@ -30,7 +30,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
- * SASL Authenitcation engine.
+ * SASL Authentication engine.
  */
 public class AmqpAuthenticator {
 
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/protocol/AmqpWireFormatTest.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/protocol/AmqpWireFormatTest.java
index a9721adeae..7e320d24d3 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/protocol/AmqpWireFormatTest.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/protocol/AmqpWireFormatTest.java
@@ -35,10 +35,10 @@ public class AmqpWireFormatTest {
         wireFormat.setAllowNonSaslConnections(false);
 
         AmqpHeader nonSaslHeader = new AmqpHeader();
-        assertFalse(wireFormat.isHeaderValid(nonSaslHeader));
+        assertFalse(wireFormat.isHeaderValid(nonSaslHeader, false));
         AmqpHeader saslHeader = new AmqpHeader();
         saslHeader.setProtocolId(3);
-        assertTrue(wireFormat.isHeaderValid(saslHeader));
+        assertTrue(wireFormat.isHeaderValid(saslHeader, false));
     }
 
     @Test
@@ -46,10 +46,21 @@ public class AmqpWireFormatTest {
         wireFormat.setAllowNonSaslConnections(true);
 
         AmqpHeader nonSaslHeader = new AmqpHeader();
-        assertTrue(wireFormat.isHeaderValid(nonSaslHeader));
+        assertTrue(wireFormat.isHeaderValid(nonSaslHeader, false));
         AmqpHeader saslHeader = new AmqpHeader();
         saslHeader.setProtocolId(3);
-        assertTrue(wireFormat.isHeaderValid(saslHeader));
+        assertTrue(wireFormat.isHeaderValid(saslHeader, false));
+    }
+
+    @Test
+    public void testNonSaslHeaderAfterSaslAuthenticationIsAccepted() {
+        wireFormat.setAllowNonSaslConnections(false);
+
+        AmqpHeader nonSaslHeader = new AmqpHeader();
+        assertTrue(wireFormat.isHeaderValid(nonSaslHeader, true));
+        AmqpHeader saslHeader = new AmqpHeader();
+        saslHeader.setProtocolId(3);
+        assertTrue(wireFormat.isHeaderValid(saslHeader, false));
     }
 
     @Test