Implement Java client SASL using javax.security.sasl, since it's actually fairly non-crazy.

Author: Simon MacMullen

src/com/rabbitmq/client/AuthMechanism.java

@@ -33,7 +33,6 @@
 import java.io.IOException;
 import java.security.KeyManagementException;
 import java.security.NoSuchAlgorithmException;
-import java.util.List;
 import java.util.Map;
 
 import java.net.Socket;
@@ -46,7 +45,6 @@
 
 import com.rabbitmq.client.impl.AMQConnection;
 import com.rabbitmq.client.impl.FrameHandler;
-import com.rabbitmq.client.impl.PlainMechanismFactory;
 import com.rabbitmq.client.impl.SocketFrameHandler;
 
 /**
@@ -86,10 +84,6 @@ public class ConnectionFactory implements Cloneable {
     /** The default port to use for AMQP connections when using SSL */
     public static final int DEFAULT_AMQP_OVER_SSL_PORT = 5671;
 
-    /** The default list of authentication mechanisms to use */
-    public static final AuthMechanismFactory[] DEFAULT_AUTH_MECHANISMS =
-        new AuthMechanismFactory[] { new PlainMechanismFactory() };
-
     /**
      * The default SSL protocol (currently "SSLv3").
      */
@@ -105,7 +99,7 @@ public class ConnectionFactory implements Cloneable {
     private int requestedHeartbeat                = DEFAULT_HEARTBEAT;
     private Map<String, Object> _clientProperties = AMQConnection.defaultClientProperties();
     private SocketFactory factory                 = SocketFactory.getDefault();
-    private AuthMechanismFactory[] authMechanismFactories = DEFAULT_AUTH_MECHANISMS;
+    private SaslConfig saslConfig                 = new DefaultSaslConfig(this);
 
     /**
      * Instantiate a ConnectionFactory with a default set of parameters.
@@ -268,29 +262,19 @@ public void setClientProperties(Map<String, Object> clientProperties) {
     }
 
     /**
-     * Given a list of mechanism names supported by the server, select a
-     * preferred mechanism, or null if we have none in common.
-     * @param serverMechanisms
+     * Gets the sasl config to use when authenticating
      * @return
      */
-    public AuthMechanismFactory getAuthMechanismFactory(List<String> serverMechanisms) {
-        // Our list is in order of preference, the server one is not.
-        for (AuthMechanismFactory f : authMechanismFactories) {
-            if (serverMechanisms.contains(f.getName())) {
-                return f;
-            }
-        }
-
-        return null;
+    public SaslConfig getSaslConfig() {
+        return saslConfig;
     }
 
     /**
-     * Set authentication mechanisms to use (in descending preference order)
-     * @param factories
-     * @see #DEFAULT_AUTH_MECHANISMS
+     * Sets the sasl config to use when authenticating
+     * @param saslConfig
      */
-    public void setAuthMechanismFactories(AuthMechanismFactory[] factories) {
-        this.authMechanismFactories = factories;
+    public void setSaslConfig(SaslConfig saslConfig) {
+        this.saslConfig = saslConfig;
     }
 
     /**

src/com/rabbitmq/client/AuthMechanismFactory.java

@@ -0,0 +1,39 @@
+package com.rabbitmq.client;
+
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.sasl.Sasl;
+import javax.security.sasl.SaslClient;
+import javax.security.sasl.SaslException;
+import java.util.Map;
+
+/**
+ *
+ */
+public class DefaultSaslConfig implements SaslConfig {
+    private ConnectionFactory factory;
+    private String authorizationId;
+    private Map<String,?> mechanismProperties;
+    private CallbackHandler callbackHandler;
+
+    public DefaultSaslConfig(ConnectionFactory factory) {
+        this.factory = factory;
+        callbackHandler = new UsernamePasswordCallbackHandler(factory);
+    }
+
+    public void setAuthorizationId(String authorizationId) {
+        this.authorizationId = authorizationId;
+    }
+
+    public void setMechanismProperties(Map<String, ?> mechanismProperties) {
+        this.mechanismProperties = mechanismProperties;
+    }
+
+    public void setCallbackHandler(CallbackHandler callbackHandler) {
+        this.callbackHandler = callbackHandler;
+    }
+
+    public SaslClient getSaslClient(String[] mechanisms) throws SaslException {
+        return Sasl.createSaslClient(mechanisms, authorizationId, "AMQP",
+              factory.getHost(), mechanismProperties, callbackHandler);
+    }
+}

src/com/rabbitmq/client/ConnectionFactory.java

@@ -0,0 +1,11 @@
+package com.rabbitmq.client;
+
+import javax.security.sasl.SaslClient;
+import javax.security.sasl.SaslException;
+
+/**
+ *
+ */
+public interface SaslConfig {
+    SaslClient getSaslClient(String[] mechanisms) throws SaslException;
+}

src/com/rabbitmq/client/DefaultSaslConfig.java

@@ -0,0 +1,32 @@
+package com.rabbitmq.client;
+
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.auth.callback.NameCallback;
+import javax.security.auth.callback.PasswordCallback;
+import javax.security.auth.callback.UnsupportedCallbackException;
+import java.io.IOException;
+
+public class UsernamePasswordCallbackHandler implements CallbackHandler {
+    private ConnectionFactory factory;
+    public UsernamePasswordCallbackHandler(ConnectionFactory factory) {
+        this.factory = factory;
+    }
+
+    public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
+        for (Callback callback: callbacks) {
+            if (callback instanceof NameCallback) {
+                NameCallback nc = (NameCallback)callback;
+                nc.setName(factory.getUsername());
+
+            } else if (callback instanceof PasswordCallback) {
+                PasswordCallback pc = (PasswordCallback)callback;
+                pc.setPassword(factory.getPassword().toCharArray());
+
+            } else {
+                throw new UnsupportedCallbackException
+                        (callback, "Unrecognized Callback");
+            }
+        }
+    }
+}

src/com/rabbitmq/client/SaslConfig.java

@@ -34,16 +34,12 @@
 import java.io.EOFException;
 import java.io.IOException;
 import java.net.SocketException;
-import java.util.Arrays;
-import java.util.List;
 import java.util.Map;
 import java.util.HashMap;
 import java.util.concurrent.TimeoutException;
 
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.AlreadyClosedException;
-import com.rabbitmq.client.AuthMechanism;
-import com.rabbitmq.client.AuthMechanismFactory;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Command;
 import com.rabbitmq.client.Connection;
@@ -53,6 +49,8 @@
 import com.rabbitmq.utility.BlockingCell;
 import com.rabbitmq.utility.Utility;
 
+import javax.security.sasl.SaslClient;
+
 /**
  * Concrete class representing and managing an AMQP connection to a broker.
  * <p>
@@ -273,22 +271,21 @@ public void start()
             throw AMQChannel.wrap(sse);
         }
 
-        List<String> mechanisms = Arrays.asList(
-                    connStart.getMechanisms().toString().split(" "));
-        AuthMechanismFactory mechanismFactory = _factory.getAuthMechanismFactory(mechanisms);
-        if (mechanismFactory == null) {
+        String[] mechanisms = connStart.getMechanisms().toString().split(" ");
+        SaslClient sc = _factory.getSaslConfig().getSaslClient(mechanisms);
+        if (sc == null) {
             throw new IOException("No compatible authentication mechanism found - " +
                     "server offered [" + connStart.getMechanisms() + "]");
         }
-        AuthMechanism mechanism = mechanismFactory.getInstance();
 
         LongString challenge = null;
+        LongString response = LongStringHelper.asLongString(
+                sc.hasInitialResponse() ? sc.evaluateChallenge(new byte[0]) : null);
         AMQP.Connection.Tune connTune = null;
         do {
-            LongString response = mechanism.handleChallenge(challenge, _factory);
             Method method = (challenge == null)
                 ? new AMQImpl.Connection.StartOk(_clientProperties,
-                                                 mechanismFactory.getName(),
+                                                 sc.getMechanismName(),
                                                  response, "en_US")
                 : new AMQImpl.Connection.SecureOk(response);
 
@@ -298,12 +295,20 @@ public void start()
                     connTune = (AMQP.Connection.Tune) serverResponse;
                 } else {
                     challenge = ((AMQP.Connection.Secure) serverResponse).getChallenge();
+                    response = LongStringHelper.asLongString(sc.evaluateChallenge(challenge.getBytes()));
                 }
             } catch (ShutdownSignalException e) {
                 throw AMQChannel.wrap(e, "Possibly caused by authentication failure");
             }
         } while (connTune == null);
 
+        sc.dispose();
+
+        if (!sc.isComplete()) {
+            throw new RuntimeException(sc.getMechanismName() +
+                    " did not complete, server thought it did");
+        }
+
         int channelMax =
             negotiatedMaxValue(_factory.getRequestedChannelMax(),
                                connTune.getChannelMax());