ARTEMIS-901 Account for the presence of authzid in sasl plan auth

If the SASL plain mechanism arrives with the authzid value set the
mechanism needs to account for its presence and use the correct fields
of the exchange to get the username and password values.  Adds some
tests to validate this fix.

artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/sasl/ServerSASLPlain.java

@@ -27,23 +27,22 @@ public class ServerSASLPlain implements ServerSASL {
 
    @Override
    public SASLResult processSASL(byte[] data) {
-
       String username = null;
       String password = null;
       String bytes = new String(data);
       String[] credentials = bytes.split(Character.toString((char) 0));
-      int offSet = 0;
-      if (credentials.length > 0) {
-         if (credentials[0].length() == 0) {
-            offSet = 1;
-         }
 
-         if (credentials.length >= offSet) {
-            username = credentials[offSet];
-         }
-         if (credentials.length >= (offSet + 1)) {
-            password = credentials[offSet + 1];
-         }
+      switch (credentials.length) {
+         case 2:
+            username = credentials[0];
+            password = credentials[1];
+            break;
+         case 3:
+            username = credentials[1];
+            password = credentials[2];
+            break;
+         default:
+            break;
       }
 
       boolean success = authenticate(username, password);

tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/AmqpConnection.java

@@ -16,6 +16,8 @@
  */
 package org.apache.activemq.transport.amqp.client;
 
+import static org.apache.activemq.transport.amqp.AmqpSupport.CONNECTION_OPEN_FAILED;
+
 import java.io.IOException;
 import java.net.URI;
 import java.nio.ByteBuffer;
@@ -30,9 +32,6 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.Unpooled;
-import io.netty.util.ReferenceCountUtil;
 import org.apache.activemq.transport.InactivityIOException;
 import org.apache.activemq.transport.amqp.client.sasl.SaslAuthenticator;
 import org.apache.activemq.transport.amqp.client.transport.NettyTransportListener;
@@ -54,7 +53,9 @@ import org.apache.qpid.proton.engine.impl.TransportImpl;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import static org.apache.activemq.transport.amqp.AmqpSupport.CONNECTION_OPEN_FAILED;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.util.ReferenceCountUtil;
 
 public class AmqpConnection extends AmqpAbstractResource<Connection> implements NettyTransportListener {
 
@@ -169,6 +170,7 @@ public class AmqpConnection extends AmqpAbstractResource<Connection> implements
             }
          });
 
+         try {
             if (connectTimeout <= 0) {
                future.sync();
             } else {
@@ -177,6 +179,14 @@ public class AmqpConnection extends AmqpAbstractResource<Connection> implements
                   throw new IOException("Failed to connect after configured timeout.");
                }
             }
+         } catch (Throwable e) {
+            try {
+               close();
+            } catch (Throwable ignore) {
+            }
+
+            throw e;
+         }
       }
    }
 

tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpSecurityTest.java

@@ -16,6 +16,12 @@
  */
 package org.apache.activemq.artemis.tests.integration.amqp;
 
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
 import org.apache.activemq.artemis.api.core.SimpleString;
 import org.apache.activemq.artemis.core.config.Configuration;
 import org.apache.activemq.artemis.core.security.Role;
@@ -33,12 +39,6 @@ import org.apache.activemq.transport.amqp.client.AmqpValidator;
 import org.apache.qpid.proton.engine.Delivery;
 import org.junit.Test;
 
-import java.io.IOException;
-import java.util.HashSet;
-import java.util.Set;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-
 public class AmqpSecurityTest extends AmqpClientTestSupport {
 
    @Override
@@ -61,6 +61,55 @@ public class AmqpSecurityTest extends AmqpClientTestSupport {
       return server;
    }
 
+   @Test(timeout = 60000)
+   public void testSaslAuthWithInvalidCredentials() throws Exception {
+      AmqpConnection connection = null;
+      AmqpClient client = createAmqpClient("foo", "foo");
+
+      try {
+         connection = client.connect();
+         fail("Should authenticate even with authzid set");
+      } catch (Exception ex) {
+      } finally {
+         if (connection != null) {
+            connection.close();
+         }
+      }
+   }
+
+   @Test(timeout = 60000)
+   public void testSaslAuthWithAuthzid() throws Exception {
+      AmqpConnection connection = null;
+      AmqpClient client = createAmqpClient("foo", "bar");
+      client.setAuthzid("foo");
+
+      try {
+         connection = client.connect();
+      } catch (Exception ex) {
+         fail("Should authenticate even with authzid set");
+      } finally {
+         if (connection != null) {
+            connection.close();
+         }
+      }
+   }
+
+   @Test(timeout = 60000)
+   public void testSaslAuthWithoutAuthzid() throws Exception {
+      AmqpConnection connection = null;
+      AmqpClient client = createAmqpClient("foo", "bar");
+
+      try {
+         connection = client.connect();
+      } catch (Exception ex) {
+         fail("Should authenticate even with authzid set");
+      } finally {
+         if (connection != null) {
+            connection.close();
+         }
+      }
+   }
+
    @Test(timeout = 60000)
    public void testSendAndRejected() throws Exception {
       AmqpConnection connection = null;