merge bug23593 into default

rade · rade · commit 65c033607dea · 2011-01-10T21:22:26.000Z
diff --git a/src/com/rabbitmq/client/Channel.java b/src/com/rabbitmq/client/Channel.java
@@ -271,7 +271,7 @@ void basicPublish(String exchange, String routingKey, boolean mandatory, boolean
     Exchange.DeclareOk exchangeDeclare(String exchange, String type, boolean durable) throws IOException;
 
     /**
-     * Declare an exchange, via an interface that allows the complete set of arguments.
+     * Declare an exchange.
      * @see com.rabbitmq.client.AMQP.Exchange.Declare
      * @see com.rabbitmq.client.AMQP.Exchange.DeclareOk
      * @param exchange the name of the exchange
@@ -285,6 +285,28 @@ void basicPublish(String exchange, String routingKey, boolean mandatory, boolean
     Exchange.DeclareOk exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete,
                                        Map<String, Object> arguments) throws IOException;
 
+    /**
+     * Declare an exchange, via an interface that allows the complete set of
+     * arguments.
+     * @see com.rabbitmq.client.AMQP.Exchange.Declare
+     * @see com.rabbitmq.client.AMQP.Exchange.DeclareOk
+     * @param exchange the name of the exchange
+     * @param type the exchange type
+     * @param durable true if we are declaring a durable exchange (the exchange will survive a server restart)
+     * @param autoDelete true if the server should delete the exchange when it is no longer in use
+     * @param internal true if the exchange is internal, i.e. can't be directly
+     * published to by a client.
+     * @param arguments other properties (construction arguments) for the exchange
+     * @return a declaration-confirm method to indicate the exchange was successfully declared
+     * @throws java.io.IOException if an error is encountered
+     */
+    Exchange.DeclareOk exchangeDeclare(String exchange,
+                                              String type,
+                                              boolean durable,
+                                              boolean autoDelete,
+                                              boolean internal,
+                                              Map<String, Object> arguments) throws IOException;
+
     /**
      * Declare an exchange passively; that is, check if the named exchange exists.
      * @param name check the existence of an exchange named this
diff --git a/src/com/rabbitmq/client/ConnectionFactory.java b/src/com/rabbitmq/client/ConnectionFactory.java
@@ -33,7 +33,6 @@
 import java.io.IOException;
 import java.security.KeyManagementException;
 import java.security.NoSuchAlgorithmException;
-import java.util.HashMap;
 import java.util.Map;
 
 import java.net.Socket;
@@ -100,6 +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 SaslConfig saslConfig                 = new DefaultSaslConfig(this);
 
     /**
      * Instantiate a ConnectionFactory with a default set of parameters.
@@ -261,6 +261,24 @@ public void setClientProperties(Map<String, Object> clientProperties) {
         _clientProperties = clientProperties;
     }
 
+    /**
+     * Gets the sasl config to use when authenticating
+     * @return
+     * @see com.rabbitmq.client.SaslConfig
+     */
+    public SaslConfig getSaslConfig() {
+        return saslConfig;
+    }
+
+    /**
+     * Sets the sasl config to use when authenticating
+     * @param saslConfig
+     * @see com.rabbitmq.client.SaslConfig
+     */
+    public void setSaslConfig(SaslConfig saslConfig) {
+        this.saslConfig = saslConfig;
+    }
+
     /**
      * Retrieve the socket factory used to make connections with.
      */
diff --git a/src/com/rabbitmq/client/DefaultSaslConfig.java b/src/com/rabbitmq/client/DefaultSaslConfig.java
@@ -0,0 +1,41 @@
+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;
+
+/**
+ * Default implementation of SaslConfig that uses the standard Java
+ * algorithm for selecting a sasl client.
+ * @see com.rabbitmq.client.ConnectionFactory
+ */
+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);
+    }
+}
diff --git a/src/com/rabbitmq/client/SaslConfig.java b/src/com/rabbitmq/client/SaslConfig.java
@@ -0,0 +1,13 @@
+package com.rabbitmq.client;
+
+import javax.security.sasl.SaslClient;
+import javax.security.sasl.SaslException;
+
+/**
+ * This interface represents a hook to allow you to control how exactly
+ * a sasl client is selected during authentication.
+ * @see com.rabbitmq.client.ConnectionFactory
+ */
+public interface SaslConfig {
+    SaslClient getSaslClient(String[] mechanisms) throws SaslException;
+}
diff --git a/src/com/rabbitmq/client/UsernamePasswordCallbackHandler.java b/src/com/rabbitmq/client/UsernamePasswordCallbackHandler.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");
+            }
+        }
+    }
+}
diff --git a/src/com/rabbitmq/client/impl/AMQConnection.java b/src/com/rabbitmq/client/impl/AMQConnection.java
@@ -51,6 +51,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>
@@ -152,7 +154,7 @@ public void ensureIsOpen()
      */
     private int _heartbeat;
 
-    private final String _username, _password, _virtualHost;
+    private final String _virtualHost;
     private final int _requestedChannelMax, _requestedFrameMax, _requestedHeartbeat;
     private final Map<String, Object> _clientProperties;
 
@@ -200,8 +202,6 @@ public AMQConnection(ConnectionFactory factory,
     {
         checkPreconditions();
 
-        _username = factory.getUsername();
-        _password = factory.getPassword();
         _virtualHost = factory.getVirtualHost();
         _requestedChannelMax = factory.getRequestedChannelMax();
         _requestedFrameMax = factory.getRequestedFrameMax();
@@ -255,8 +255,9 @@ public void start()
         ml.setName("AMQP Connection " + getHost() + ":" + getPort());
         ml.start();
 
+        AMQP.Connection.Start connStart = null;
         try {
-            AMQP.Connection.Start connStart =
+            connStart =
                 (AMQP.Connection.Start) connStartBlocker.getReply().getMethod();
 
             _serverProperties = connStart.getServerProperties();
@@ -274,18 +275,42 @@ public void start()
             throw AMQChannel.wrap(sse);
         }
 
-        LongString saslResponse = LongStringHelper.asLongString("\0" + _username +
-                                                                "\0" + _password);
-        AMQImpl.Connection.StartOk startOk =
-            new AMQImpl.Connection.StartOk(_clientProperties, "PLAIN",
-                                           saslResponse, "en_US");
+        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() + "]");
+        }
 
+        LongString challenge = null;
+        LongString response = LongStringHelper.asLongString(
+                sc.hasInitialResponse() ? sc.evaluateChallenge(new byte[0]) : null);
         AMQP.Connection.Tune connTune = null;
+        do {
+            Method method = (challenge == null)
+                ? new AMQImpl.Connection.StartOk(_clientProperties,
+                                                 sc.getMechanismName(),
+                                                 response, "en_US")
+                : new AMQImpl.Connection.SecureOk(response);
 
-        try {
-            connTune = (AMQP.Connection.Tune) _channel0.rpc(startOk).getMethod();
-        } catch (ShutdownSignalException e) {
-            throw new PossibleAuthenticationFailureException(e);
+            try {
+                Method serverResponse = _channel0.rpc(method).getMethod();
+                if (serverResponse instanceof AMQP.Connection.Tune) {
+                    connTune = (AMQP.Connection.Tune) serverResponse;
+                } else {
+                    challenge = ((AMQP.Connection.Secure) serverResponse).getChallenge();
+                    response = LongStringHelper.asLongString(sc.evaluateChallenge(challenge.getBytes()));
+                }
+            } catch (ShutdownSignalException e) {
+                throw new PossibleAuthenticationFailureException(e);
+            }
+        } while (connTune == null);
+
+        sc.dispose();
+
+        if (!sc.isComplete()) {
+            throw new RuntimeException(sc.getMechanismName() +
+                    " did not complete, server thought it did");
         }
 
         int channelMax =
@@ -714,6 +739,6 @@ public void close(int closeCode,
     }
 
     @Override public String toString() {
-        return "amqp://" + _username + "@" + getHost() + ":" + getPort() + _virtualHost;
+        return "amqp://" + _factory.getUsername() + "@" + getHost() + ":" + getPort() + _virtualHost;
     }
 }
diff --git a/src/com/rabbitmq/client/impl/CRDemoSaslClient.java b/src/com/rabbitmq/client/impl/CRDemoSaslClient.java
@@ -0,0 +1,103 @@
+package com.rabbitmq.client.impl;
+
+import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.SaslConfig;
+import com.rabbitmq.client.UsernamePasswordCallbackHandler;
+
+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 javax.security.sasl.SaslClient;
+import javax.security.sasl.SaslException;
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.util.Arrays;
+
+/**
+    Provides equivalent security to PLAIN but demos use of Connection.Secure(Ok)
+    START-OK: Username
+    SECURE: "Please tell me your password"
+    SECURE-OK: Password
+*/
+
+public class CRDemoSaslClient implements SaslClient {
+    private static final String NAME = "RABBIT-CR-DEMO";
+
+    private CallbackHandler handler;
+    private int round = 0;
+
+    public CRDemoSaslClient(CallbackHandler handler) {
+        this.handler = handler;
+    }
+
+    public String getMechanismName() {
+        return NAME;
+    }
+
+    public boolean hasInitialResponse() {
+        return true;
+    }
+
+    public byte[] evaluateChallenge(byte[] challenge) throws SaslException {
+        byte[] resp;
+        try {
+            if (round == 0) {
+                NameCallback nc = new NameCallback("Name:");
+                handler.handle(new Callback[]{nc});
+                resp = nc.getName().getBytes("utf-8");
+            } else {
+                PasswordCallback pc = new PasswordCallback("Password:", false);
+                handler.handle(new Callback[]{pc});
+                resp = ("My password is " + new String(pc.getPassword())).getBytes("utf-8");
+            }
+        } catch (UnsupportedEncodingException e) {
+            throw new RuntimeException(e);
+        } catch (UnsupportedCallbackException e) {
+            throw new SaslException("Bad callback", e);
+        } catch (IOException e) {
+            throw new SaslException("IO Exception", e);
+        }
+
+        round++;
+        return resp;
+    }
+
+    public boolean isComplete() {
+        return round == 2;
+    }
+
+    public byte[] unwrap(byte[] bytes, int i, int i1) throws SaslException {
+        throw new UnsupportedOperationException();
+    }
+
+    public byte[] wrap(byte[] bytes, int i, int i1) throws SaslException {
+        throw new UnsupportedOperationException();
+    }
+
+    public Object getNegotiatedProperty(String s) {
+        throw new UnsupportedOperationException();
+    }
+
+    public void dispose() throws SaslException {
+        // NOOP
+    }
+
+    public static class CRDemoSaslConfig implements SaslConfig {
+        private ConnectionFactory factory;
+
+        public CRDemoSaslConfig(ConnectionFactory factory) {
+            this.factory = factory;
+        }
+
+        public SaslClient getSaslClient(String[] mechanisms) throws SaslException {
+            if (Arrays.asList(mechanisms).contains(NAME)) {
+                return new CRDemoSaslClient(new UsernamePasswordCallbackHandler(factory));
+            }
+            else {
+                return null;
+            }
+        }
+    }
+}
diff --git a/src/com/rabbitmq/client/impl/ChannelN.java b/src/com/rabbitmq/client/impl/ChannelN.java
@@ -514,11 +514,25 @@ public Exchange.DeclareOk exchangeDeclare(String exchange, String type,
                                               boolean durable, boolean autoDelete,
                                               Map<String, Object> arguments)
         throws IOException
+    {
+        return exchangeDeclare(exchange, type,
+                               durable, autoDelete, false,
+                               arguments);
+    }
+
+    /** Public API - {@inheritDoc} */
+    public Exchange.DeclareOk exchangeDeclare(String exchange, String type,
+                                              boolean durable,
+                                              boolean autoDelete,
+                                              boolean internal,
+                                              Map<String, Object> arguments)
+            throws IOException
     {
         return (Exchange.DeclareOk)
-            exnWrappingRpc(new Exchange.Declare(TICKET, exchange, type,
-                                                false, durable, autoDelete,
-                                                false, false, arguments)).getMethod();
+                exnWrappingRpc(new Exchange.Declare(TICKET, exchange, type,
+                                                    false, durable, autoDelete,
+                                                    internal, false,
+                                                    arguments)).getMethod();
     }
 
     /** Public API - {@inheritDoc} */
diff --git a/test/src/com/rabbitmq/client/test/functional/FunctionalTests.java b/test/src/com/rabbitmq/client/test/functional/FunctionalTests.java
diff --git a/test/src/com/rabbitmq/client/test/functional/InternalExchange.java b/test/src/com/rabbitmq/client/test/functional/InternalExchange.java
diff --git a/test/src/com/rabbitmq/client/test/functional/SaslMechanisms.java b/test/src/com/rabbitmq/client/test/functional/SaslMechanisms.java
diff --git a/test/src/com/rabbitmq/client/test/functional/UserIDHeader.java b/test/src/com/rabbitmq/client/test/functional/UserIDHeader.java
