AMQ-6055: uptest test client and add [currently-ignored] test to demonstrate the issue

activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpClient.java

@@ -38,6 +38,8 @@ public class AmqpClient {
     private final String username;
     private final String password;
     private final URI remoteURI;
+    private String authzid;
+    private String mechanismRestriction;
 
     private AmqpValidator stateInspector = new AmqpValidator();
     private List<Symbol> offeredCapabilities = Collections.emptyList();
@@ -95,6 +97,9 @@ public class AmqpClient {
         NettyTransport transport = NettyTransportFactory.createTransport(remoteURI);
         AmqpConnection connection = new AmqpConnection(transport, username, password);
 
+        connection.setMechanismRestriction(mechanismRestriction);
+        connection.setAuthzid(authzid);
+
         connection.setOfferedCapabilities(getOfferedCapabilities());
         connection.setOfferedProperties(getOfferedProperties());
         connection.setStateInspector(getStateInspector());
@@ -103,19 +108,43 @@ public class AmqpClient {
     }
 
     /**
-     * @return the user name value given when connect was called, always null before connect.
+     * @return the user name value given when constructed.
      */
     public String getUsername() {
         return username;
     }
 
     /**
-     * @return the password value given when connect was called, always null before connect.
+     * @return the password value given when constructed.
      */
     public String getPassword() {
         return password;
     }
 
+    /**
+     * @param authzid
+     *        The authzid used when authenticating (currently only with PLAIN)
+     */
+    public void setAuthzid(String authzid) {
+        this.authzid = authzid;
+    }
+
+    public String getAuthzid() {
+        return authzid;
+    }
+
+    /**
+     * @param mechanismRestriction
+     *        The mechanism to use when authenticating (if offered by the server)
+     */
+    public void setMechanismRestriction(String mechanismRestriction) {
+        this.mechanismRestriction = mechanismRestriction;
+    }
+
+    public String getMechanismRestriction() {
+        return mechanismRestriction;
+    }
+
     /**
      * @return the currently set address to use to connect to the AMQP peer.
      */

activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/AmqpConnection.java

@@ -84,6 +84,8 @@ public class AmqpConnection extends AmqpAbstractResource<Connection> implements
 
     private AmqpConnectionListener listener;
     private SaslAuthenticator authenticator;
+    private String mechanismRestriction;
+    private String authzid;
 
     private int idleTimeout = 0;
     private boolean idleProcessingDisabled;
@@ -144,7 +146,7 @@ public class AmqpConnection extends AmqpAbstractResource<Connection> implements
                     if (sasl != null) {
                         sasl.client();
                     }
-                    authenticator = new SaslAuthenticator(sasl, username, password);
+                    authenticator = new SaslAuthenticator(sasl, username, password, authzid, mechanismRestriction);
                     open(future);
 
                     pumpToProtonTransport();
@@ -288,6 +290,14 @@ public class AmqpConnection extends AmqpAbstractResource<Connection> implements
         return password;
     }
 
+    public void setAuthzid(String authzid) {
+        this.authzid = authzid;
+    }
+
+    public String getAuthzid() {
+        return authzid;
+    }
+
     /**
      * @return the URI of the remote peer this connection attached to.
      */
@@ -396,6 +406,19 @@ public class AmqpConnection extends AmqpAbstractResource<Connection> implements
         return idleProcessingDisabled;
     }
 
+    /**
+     * Sets a restriction on the SASL mechanism to use (if offered by the server).
+     *
+     * @param mechanismRestriction the mechanism to use
+     */
+    public void setMechanismRestriction(String mechanismRestriction) {
+        this.mechanismRestriction = mechanismRestriction;
+    }
+
+    public String getMechanismRestriction() {
+        return mechanismRestriction;
+    }
+
     //----- Internal getters used from the child AmqpResource classes --------//
 
     ScheduledExecutorService getScheduler() {

activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/sasl/AbstractMechanism.java

@@ -29,6 +29,7 @@ public abstract class AbstractMechanism implements Mechanism {
 
     private String username;
     private String password;
+    private String authzid;
     private Map<String, Object> properties = new HashMap<String, Object>();
 
     @Override
@@ -77,4 +78,14 @@ public abstract class AbstractMechanism implements Mechanism {
     public String toString() {
         return "SASL-" + getName();
     }
+
+    @Override
+    public String getAuthzid() {
+        return authzid;
+    }
+
+    @Override
+    public void setAuthzid(String authzid) {
+        this.authzid = authzid;
+    }
 }

activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/sasl/Mechanism.java

@@ -122,4 +122,7 @@ public interface Mechanism extends Comparable<Mechanism> {
      */
     Map<String, Object> getProperties();
 
+    String getAuthzid();
+
+    void setAuthzid(String authzid);
 }

activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/sasl/PlainMechanism.java

@@ -23,6 +23,8 @@ package org.apache.activemq.transport.amqp.client.sasl;
  */
 public class PlainMechanism extends AbstractMechanism {
 
+    public static final String MECH_NAME = "PLAIN";
+
     @Override
     public int getPriority() {
         return PRIORITY.MEDIUM.getValue();
@@ -30,15 +32,20 @@ public class PlainMechanism extends AbstractMechanism {
 
     @Override
     public String getName() {
-        return "PLAIN";
+        return MECH_NAME;
     }
 
     @Override
     public byte[] getInitialResponse() {
 
+        String authzid = getAuthzid();
         String username = getUsername();
         String password = getPassword();
 
+        if (authzid == null) {
+            authzid = "";
+        }
+
         if (username == null) {
             username = "";
         }
@@ -47,10 +54,12 @@ public class PlainMechanism extends AbstractMechanism {
             password = "";
         }
 
+        byte[] authzidBytes = authzid.getBytes();
         byte[] usernameBytes = username.getBytes();
         byte[] passwordBytes = password.getBytes();
-        byte[] data = new byte[usernameBytes.length + passwordBytes.length + 2];
+        byte[] data = new byte[authzidBytes.length + 1 + usernameBytes.length + 1 + passwordBytes.length];
-        System.arraycopy(usernameBytes, 0, data, 1, usernameBytes.length);
+        System.arraycopy(authzidBytes, 0, data, 0, authzidBytes.length);
+        System.arraycopy(usernameBytes, 0, data, 1 + authzidBytes.length, usernameBytes.length);
         System.arraycopy(passwordBytes, 0, data, 2 + usernameBytes.length, passwordBytes.length);
         return data;
     }

activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/client/sasl/SaslAuthenticator.java

@@ -37,7 +37,9 @@ public class SaslAuthenticator {
     private final Sasl sasl;
     private final String username;
     private final String password;
+    private final String authzid;
     private Mechanism mechanism;
+    private String mechanismRestriction;
 
     /**
      * Create the authenticator and initialize it.
@@ -48,11 +50,17 @@ public class SaslAuthenticator {
      *        The user name that will be used to authenticate.
      * @param password
      *        The password that will be used to authenticate.
+     * @param authzid
+     *        The authzid used when authenticating (currently only with PLAIN)
+     * @param mechanismRestriction
+     *        A particular mechanism to use (if offered by the server) or null to allow selection.
      */
-    public SaslAuthenticator(Sasl sasl, String username, String password) {
+    public SaslAuthenticator(Sasl sasl, String username, String password, String authzid, String mechanismRestriction) {
         this.sasl = sasl;
         this.username = username;
         this.password = password;
+        this.authzid = authzid;
+        this.mechanismRestriction = mechanismRestriction;
     }
 
     /**
@@ -90,6 +98,7 @@ public class SaslAuthenticator {
                 if (mechanism != null) {
                     mechanism.setUsername(username);
                     mechanism.setPassword(password);
+                    mechanism.setAuthzid(authzid);
                     // TODO - set additional options from URI.
                     // TODO - set a host value.
 
@@ -117,6 +126,11 @@ public class SaslAuthenticator {
         List<Mechanism> found = new ArrayList<Mechanism>();
 
         for (String remoteMechanism : remoteMechanisms) {
+            if(mechanismRestriction != null && !mechanismRestriction.equals(remoteMechanism)) {
+                LOG.debug("Skipping {} mechanism because it is not the configured mechanism restriction {}", remoteMechanism, mechanismRestriction);
+                continue;
+            }
+
             if (remoteMechanism.equalsIgnoreCase("PLAIN")) {
                 found.add(new PlainMechanism());
             } else if (remoteMechanism.equalsIgnoreCase("ANONYMOUS")) {

activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpSaslPlainTest.java

@@ -0,0 +1,127 @@
+/*
+ * 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.activemq.transport.amqp.interop;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.activemq.broker.BrokerPlugin;
+import org.apache.activemq.security.AuthenticationUser;
+import org.apache.activemq.security.SimpleAuthenticationPlugin;
+import org.apache.activemq.transport.amqp.client.AmqpClient;
+import org.apache.activemq.transport.amqp.client.AmqpClientTestSupport;
+import org.apache.activemq.transport.amqp.client.AmqpConnection;
+import org.apache.activemq.transport.amqp.client.AmqpSender;
+import org.apache.activemq.transport.amqp.client.AmqpSession;
+import org.apache.activemq.transport.amqp.client.sasl.PlainMechanism;
+import org.junit.Ignore;
+import org.junit.Test;
+
+/**
+ * Test broker behaviour when creating AMQP connections with SASL PLAIN mechanism.
+ */
+public class AmqpSaslPlainTest extends AmqpClientTestSupport {
+
+    private static final String ADMIN = "admin";
+    private static final String USER = "user";
+    private static final String USER_PASSWORD = "password";
+
+    @Override
+    protected void performAdditionalConfiguration(org.apache.activemq.broker.BrokerService brokerService) throws Exception {
+        List<AuthenticationUser> users = new ArrayList<AuthenticationUser>();
+        users.add(new AuthenticationUser(USER, USER_PASSWORD, "users"));
+        users.add(new AuthenticationUser(ADMIN, ADMIN, "admins"));
+
+        SimpleAuthenticationPlugin authenticationPlugin = new SimpleAuthenticationPlugin(users);
+
+        brokerService.setPlugins(new BrokerPlugin[] { authenticationPlugin});
+    };
+
+    @Test(timeout = 30000)
+    public void testSaslPlainWithValidUsernameAndPassword() throws Exception {
+        AmqpClient client = createAmqpClient(USER, USER_PASSWORD);
+
+        doSucessfullConnectionTestImpl(client);
+    }
+
+    @Ignore //TODO: fix broker to handle authzid
+    @Test(timeout = 30000)
+    public void testSaslPlainWithValidUsernameAndPasswordAndAuthzid() throws Exception {
+        AmqpClient client = createAmqpClient(USER, USER_PASSWORD);
+        client.setAuthzid(USER);
+
+        doSucessfullConnectionTestImpl(client);
+    }
+
+    private void doSucessfullConnectionTestImpl(AmqpClient client) throws Exception {
+        client.setMechanismRestriction(PlainMechanism.MECH_NAME);
+
+        // Expect connection to succeed
+        AmqpConnection connection = client.connect();
+
+        // Exercise it for verification
+        exerciseConnection(connection);
+
+        connection.close();
+    }
+
+    private void exerciseConnection(AmqpConnection connection)throws Exception{
+        AmqpSession session = connection.createSession();
+
+        assertEquals(0, brokerService.getAdminView().getQueues().length);
+
+        AmqpSender sender = session.createSender("queue://" + getTestName());
+
+        assertEquals(1, brokerService.getAdminView().getQueues().length);
+        assertNotNull(getProxyToQueue(getTestName()));
+        assertEquals(1, brokerService.getAdminView().getQueueProducers().length);
+        sender.close();
+        assertEquals(0, brokerService.getAdminView().getQueueProducers().length);
+    }
+
+    @Test(timeout = 30000)
+    public void testSaslPlainWithInvalidUsername() throws Exception {
+        AmqpClient client = createAmqpClient("not-user", USER_PASSWORD);
+        doFailedConnectionTestImpl(client);
+    }
+
+    @Test(timeout = 30000)
+    public void testSaslPlainWithInvalidPassword() throws Exception {
+        AmqpClient client = createAmqpClient(USER, "not-user-password");
+        doFailedConnectionTestImpl(client);
+    }
+
+    private void doFailedConnectionTestImpl(AmqpClient client) throws Exception {
+        client.setMechanismRestriction(PlainMechanism.MECH_NAME);
+
+        // Expect connection to fail
+        try {
+            client.connect();
+            fail("exected connection to fail");
+        } catch (Exception e){
+            // Expected
+            Throwable cause = e.getCause();
+            assertNotNull("Expected security exception cause", cause);
+            assertTrue("Expected security exception cause", cause instanceof SecurityException);
+        }
+    }
+}