https://issues.apache.org/jira/browse/AMQ-6055

Account for Authzid in SASL PLAIN mechanism and provide a means to fail
the authorization if the challenge response is invalid.  Update the
client to properly exclude sasl mechanism that don't apply to it's
configured credentials such as using only ANONYMOUS when no user or
password is set.

activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/sasl/AbstractSaslMechanism.java

@@ -23,6 +23,8 @@ public abstract class AbstractSaslMechanism implements SaslMechanism {
 
     protected String username;
     protected String password;
+    protected boolean failed;
+    protected String failureReason;
 
     @Override
     public String getUsername() {
@@ -33,4 +35,19 @@ public abstract class AbstractSaslMechanism implements SaslMechanism {
     public String getPassword() {
         return password;
     }
+
+    @Override
+    public boolean isFailed() {
+        return failed;
+    }
+
+    @Override
+    public String getFailureReason() {
+        return failureReason;
+    }
+
+    void setFailed(String failureReason) {
+        this.failed = true;
+        this.failureReason = failureReason;
+    }
 }

activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/sasl/AmqpAuthenticator.java

@@ -75,17 +75,22 @@ public class AmqpAuthenticator {
                 LOG.debug("SASL [{}} Handshake started.", mechanism.getMechanismName());
 
                 mechanism.processSaslStep(sasl);
+                if (!mechanism.isFailed()) {
 
-                connectionInfo.setUserName(mechanism.getUsername());
+                    connectionInfo.setUserName(mechanism.getUsername());
-                connectionInfo.setPassword(mechanism.getPassword());
+                    connectionInfo.setPassword(mechanism.getPassword());
 
-                if (tryAuthenticate(connectionInfo, transport.getPeerCertificates())) {
+                    if (tryAuthenticate(connectionInfo, transport.getPeerCertificates())) {
-                    sasl.done(Sasl.SaslOutcome.PN_SASL_OK);
+                        sasl.done(Sasl.SaslOutcome.PN_SASL_OK);
+                    } else {
+                        sasl.done(Sasl.SaslOutcome.PN_SASL_AUTH);
+                    }
+
+                    LOG.debug("SASL [{}} Handshake complete.", mechanism.getMechanismName());
                 } else {
+                    LOG.debug("SASL [{}} Handshake failed: {}", mechanism.getMechanismName(), mechanism.getFailureReason());
                     sasl.done(Sasl.SaslOutcome.PN_SASL_AUTH);
                 }
-
-                LOG.debug("SASL [{}} Handshake complete.", mechanism.getMechanismName());
             } else {
                 LOG.info("SASL: could not find supported mechanism");
                 sasl.done(Sasl.SaslOutcome.PN_SASL_PERM);

activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/sasl/AnonymousMechanism.java

@@ -21,7 +21,7 @@ import org.apache.qpid.proton.engine.Sasl;
 /**
  * SASL Anonymous mechanism implementation.
  */
-public class AnonymousMechanism implements SaslMechanism {
+public class AnonymousMechanism extends AbstractSaslMechanism {
 
     @Override
     public void processSaslStep(Sasl sasl) {
@@ -31,14 +31,4 @@ public class AnonymousMechanism implements SaslMechanism {
     public String getMechanismName() {
         return "ANONYMOUS";
     }
-
-    @Override
-    public String getUsername() {
-        return null;
-    }
-
-    @Override
-    public String getPassword() {
-        return null;
-    }
 }

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

@@ -28,14 +28,25 @@ public class PlainMechanism extends AbstractSaslMechanism {
     public void processSaslStep(Sasl sasl) {
         byte[] data = new byte[sasl.pending()];
         sasl.recv(data, 0, data.length);
+
         Buffer[] parts = new Buffer(data).split((byte) 0);
-
+        switch (parts.length) {
-        if (parts.length > 0) {
+            case 0:
-            username = parts[0].utf8().toString();
+                // Treat this as anonymous connect to support legacy behavior
-        }
+                // which allowed this.  Connection will fail if broker is not
-
+                // configured to allow anonymous connections.
-        if (parts.length > 1) {
+                break;
-            password = parts[1].utf8().toString();
+            case 2:
+                username = parts[0].utf8().toString();
+                password = parts[1].utf8().toString();
+                break;
+            case 3:
+                username = parts[1].utf8().toString();
+                password = parts[2].utf8().toString();
+                break;
+            default:
+                setFailed("Invalid encoding of Authentication credentials");
+                break;
         }
     }
 

activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/sasl/SaslMechanism.java

@@ -48,4 +48,14 @@ public interface SaslMechanism {
      */
     String getMechanismName();
 
+    /**
+     * @return true if the SASL processing failed during a step.
+     */
+    boolean isFailed();
+
+    /**
+     * @return a failure error to explain why the mechanism failed.
+     */
+    String getFailureReason();
+
 }

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

@@ -88,4 +88,9 @@ public abstract class AbstractMechanism implements Mechanism {
     public void setAuthzid(String authzid) {
         this.authzid = authzid;
     }
+
+    @Override
+    public boolean isApplicable(String username, String password) {
+        return true;
+    }
 }

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

@@ -83,4 +83,9 @@ public class CramMD5Mechanism extends AbstractMechanism {
             return EMPTY;
         }
     }
+
+    @Override
+    public boolean isApplicable(String username, String password) {
+        return username != null && username.length() > 0 && password != null && password.length() > 0;
+    }
 }

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

@@ -122,7 +122,32 @@ public interface Mechanism extends Comparable<Mechanism> {
      */
     Map<String, Object> getProperties();
 
+    /**
+     * Using the configured credentials, check if the mechanism applies or not.
+     *
+     * @param username
+     *      The user name that will be used with this mechanism
+     * @param password
+     *      The password that will be used with this mechanism
+     *
+     * @return true if the mechanism works with the provided credentials or not.
+     */
+    boolean isApplicable(String username, String password);
+
+    /**
+     * Get the currently configured Authentication ID.
+     *
+     * @return the currently set Authentication ID.
+     */
     String getAuthzid();
 
+    /**
+     * Sets an Authentication ID that some mechanism can use during the
+     * challenge response phase.
+     *
+     * @param authzid
+     *      The Authentication ID to use.
+     */
     void setAuthzid(String authzid);
+
 }

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

@@ -68,4 +68,9 @@ public class PlainMechanism extends AbstractMechanism {
     public byte[] getChallengeResponse(byte[] challenge) {
         return EMPTY;
     }
+
+    @Override
+    public boolean isApplicable(String username, String password) {
+        return username != null && username.length() > 0 && password != null && password.length() > 0;
+    }
 }

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

@@ -131,14 +131,20 @@ public class SaslAuthenticator {
                 continue;
             }
 
+            Mechanism mechanism = null;
             if (remoteMechanism.equalsIgnoreCase("PLAIN")) {
-                found.add(new PlainMechanism());
+                mechanism = new PlainMechanism();
             } else if (remoteMechanism.equalsIgnoreCase("ANONYMOUS")) {
-                found.add(new AnonymousMechanism());
+                mechanism = new AnonymousMechanism();
             } else if (remoteMechanism.equalsIgnoreCase("CRAM-MD5")) {
-                found.add(new CramMD5Mechanism());
+                mechanism = new CramMD5Mechanism();
             } else {
                 LOG.debug("Unknown remote mechanism {}, skipping", remoteMechanism);
+                continue;
+            }
+
+            if (mechanism.isApplicable(username, password)) {
+                found.add(mechanism);
             }
         }
 

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

@@ -33,7 +33,6 @@ 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;
 
 /**
@@ -63,15 +62,22 @@ public class AmqpSaslPlainTest extends AmqpClientTestSupport {
         doSucessfullConnectionTestImpl(client);
     }
 
-    @Ignore //TODO: fix broker to handle authzid
     @Test(timeout = 30000)
-    public void testSaslPlainWithValidUsernameAndPasswordAndAuthzid() throws Exception {
+    public void testSaslPlainWithValidUsernameAndPasswordAndAuthzidAsUser() throws Exception {
         AmqpClient client = createAmqpClient(USER, USER_PASSWORD);
         client.setAuthzid(USER);
 
         doSucessfullConnectionTestImpl(client);
     }
 
+    @Test(timeout = 30000)
+    public void testSaslPlainWithValidUsernameAndPasswordAndAuthzidAsUnkown() throws Exception {
+        AmqpClient client = createAmqpClient(USER, USER_PASSWORD);
+        client.setAuthzid("unknown");
+
+        doSucessfullConnectionTestImpl(client);
+    }
+
     private void doSucessfullConnectionTestImpl(AmqpClient client) throws Exception {
         client.setMechanismRestriction(PlainMechanism.MECH_NAME);
 
@@ -110,6 +116,20 @@ public class AmqpSaslPlainTest extends AmqpClientTestSupport {
         doFailedConnectionTestImpl(client);
     }
 
+    @Test(timeout = 30000)
+    public void testSaslPlainWithInvalidUsernameAndAuthzid() throws Exception {
+        AmqpClient client = createAmqpClient("not-user", USER_PASSWORD);
+        client.setAuthzid(USER);
+        doFailedConnectionTestImpl(client);
+    }
+
+    @Test(timeout = 30000)
+    public void testSaslPlainWithInvalidPasswordAndAuthzid() throws Exception {
+        AmqpClient client = createAmqpClient(USER, "not-user-password");
+        client.setAuthzid(USER);
+        doFailedConnectionTestImpl(client);
+    }
+
     private void doFailedConnectionTestImpl(AmqpClient client) throws Exception {
         client.setMechanismRestriction(PlainMechanism.MECH_NAME);