QPID-3415: Change 0-10 code path to utilise the CallbackHandlerRegistry to create the correct CallbackHandler. The sasl_mechs property/broker option is retained, but continues to be understood only by the 0-10 path.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1172506 13f79535-47bb-0310-9956-ffa450edef68

6 files changed, 550 insertions, 125 deletions

diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java b/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java
index 63342bdb26..0ed3db6ecb 100644
--- a/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java
+++ b/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java
@@ -35,6 +35,7 @@ import javax.jms.XASession;
 import org.apache.qpid.AMQException;
 import org.apache.qpid.client.failover.FailoverException;
 import org.apache.qpid.client.failover.FailoverProtectedOperation;
+import org.apache.qpid.client.transport.ClientConnectionDelegate;
 import org.apache.qpid.configuration.ClientProperties;
 import org.apache.qpid.framing.ProtocolVersion;
 import org.apache.qpid.jms.BrokerDetails;
@@ -194,6 +195,7 @@ public class AMQConnectionDelegate_0_10 implements AMQConnectionDelegate, Connec
             }
 
             ConnectionSettings conSettings = retriveConnectionSettings(brokerDetail);
+            _qpidConnection.setConnectionDelegate(new ClientConnectionDelegate(conSettings, _conn.getConnectionURL()));
             _qpidConnection.connect(conSettings);
 
             _conn._connected = true;
diff --git a/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionStartMethodHandler.java b/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionStartMethodHandler.java
index 2b49bb8f81..939bd181a3 100644
--- a/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionStartMethodHandler.java
+++ b/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionStartMethodHandler.java
@@ -20,6 +20,13 @@
  */
 package org.apache.qpid.client.handler;
 
+import java.io.UnsupportedEncodingException;
+import java.util.StringTokenizer;
+
+import javax.security.sasl.Sasl;
+import javax.security.sasl.SaslClient;
+import javax.security.sasl.SaslException;
+
 import org.apache.qpid.AMQException;
 import org.apache.qpid.client.protocol.AMQProtocolSession;
 import org.apache.qpid.client.security.AMQCallbackHandler;
@@ -34,18 +41,9 @@ import org.apache.qpid.framing.ConnectionStartOkBody;
 import org.apache.qpid.framing.FieldTable;
 import org.apache.qpid.framing.FieldTableFactory;
 import org.apache.qpid.framing.ProtocolVersion;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import javax.security.sasl.Sasl;
-import javax.security.sasl.SaslClient;
-import javax.security.sasl.SaslException;
-
-import java.io.UnsupportedEncodingException;
-import java.util.HashSet;
-import java.util.StringTokenizer;
-
 public class ConnectionStartMethodHandler implements StateAwareMethodListener<ConnectionStartBody>
 {
     private static final Logger _log = LoggerFactory.getLogger(ConnectionStartMethodHandler.class);
@@ -197,40 +195,20 @@ public class ConnectionStartMethodHandler implements StateAwareMethodListener<Co
     private String chooseMechanism(byte[] availableMechanisms) throws UnsupportedEncodingException
     {
         final String mechanisms = new String(availableMechanisms, "utf8");
-        StringTokenizer tokenizer = new StringTokenizer(mechanisms, " ");
-        HashSet mechanismSet = new HashSet();
-        while (tokenizer.hasMoreTokens())
-        {
-            mechanismSet.add(tokenizer.nextToken());
-        }
-
-        String preferredMechanisms = CallbackHandlerRegistry.getInstance().getMechanisms();
-        StringTokenizer prefTokenizer = new StringTokenizer(preferredMechanisms, " ");
-        while (prefTokenizer.hasMoreTokens())
-        {
-            String mech = prefTokenizer.nextToken();
-            if (mechanismSet.contains(mech))
-            {
-                return mech;
-            }
-        }
-
-        return null;
+        return CallbackHandlerRegistry.getInstance().selectMechanism(mechanisms);
     }
 
     private AMQCallbackHandler createCallbackHandler(String mechanism, AMQProtocolSession protocolSession)
         throws AMQException
     {
-        Class mechanismClass = CallbackHandlerRegistry.getInstance().getCallbackHandlerClass(mechanism);
         try
         {
-            Object instance = mechanismClass.newInstance();
-            AMQCallbackHandler cbh = (AMQCallbackHandler) instance;
-            cbh.initialise(protocolSession.getAMQConnection().getConnectionURL());
+            AMQCallbackHandler instance = CallbackHandlerRegistry.getInstance().createCallbackHandler(mechanism);
+            instance.initialise(protocolSession.getAMQConnection().getConnectionURL());
 
-            return cbh;
+            return instance;
         }
-        catch (Exception e)
+        catch (IllegalArgumentException e)
         {
             throw new AMQException(null, "Unable to create callback handler: " + e, e);
         }
diff --git a/java/client/src/main/java/org/apache/qpid/client/security/CallbackHandlerRegistry.java b/java/client/src/main/java/org/apache/qpid/client/security/CallbackHandlerRegistry.java
index 140cbdeb75..14bae68561 100644
--- a/java/client/src/main/java/org/apache/qpid/client/security/CallbackHandlerRegistry.java
+++ b/java/client/src/main/java/org/apache/qpid/client/security/CallbackHandlerRegistry.java
@@ -20,17 +20,22 @@
  */
 package org.apache.qpid.client.security;
 
-import org.apache.qpid.util.FileUtils;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.Collection;
+import java.util.Collections;
 import java.util.Enumeration;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Map;
 import java.util.Properties;
+import java.util.Set;
+import java.util.StringTokenizer;
+import java.util.TreeMap;
+
+import org.apache.qpid.util.FileUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * CallbackHandlerRegistry is a registry for call back handlers for user authentication and interaction during user
@@ -42,7 +47,7 @@ import java.util.Properties;
  * "amp.callbackhandler.properties". The format of the properties file is:
  *
  * <p/><pre>
- * CallbackHanlder.mechanism=fully.qualified.class.name
+ * CallbackHanlder.n.mechanism=fully.qualified.class.name where n is an ordinal
  * </pre>
  *
  * <p/>Where mechanism is an IANA-registered mechanism name and the fully qualified class name refers to a
@@ -66,51 +71,15 @@ public class CallbackHandlerRegistry
     public static final String DEFAULT_RESOURCE_NAME = "org/apache/qpid/client/security/CallbackHandlerRegistry.properties";
 
     /** A static reference to the singleton instance of this registry. */
-    private static CallbackHandlerRegistry _instance = new CallbackHandlerRegistry();
+    private static final CallbackHandlerRegistry _instance;
 
     /** Holds a map from SASL mechanism names to call back handlers. */
-    private Map<String, Class> _mechanismToHandlerClassMap = new HashMap<String, Class>();
-
-    /** Holds a space delimited list of mechanisms that callback handlers exist for. */
-    private String _mechanisms;
-
-    /**
-     * Gets the singleton instance of this registry.
-     *
-     * @return The singleton instance of this registry.
-     */
-    public static CallbackHandlerRegistry getInstance()
-    {
-        return _instance;
-    }
+    private Map<String, Class<AMQCallbackHandler>> _mechanismToHandlerClassMap = new HashMap<String, Class<AMQCallbackHandler>>();
 
-    /**
-     * Gets the callback handler class for a given SASL mechanism name.
-     *
-     * @param mechanism The SASL mechanism name.
-     *
-     * @return The callback handler class for the mechanism, or null if none is configured for that mechanism.
-     */
-    public Class getCallbackHandlerClass(String mechanism)
-    {
-        return (Class) _mechanismToHandlerClassMap.get(mechanism);
-    }
+    /** Ordered collection of mechanisms for which callback handlers exist. */
+    private Collection<String> _mechanisms;
 
-    /**
-     * Gets a space delimited list of supported SASL mechanisms.
-     *
-     * @return A space delimited list of supported SASL mechanisms.
-     */
-    public String getMechanisms()
-    {
-        return _mechanisms;
-    }
-
-    /**
-     * Creates the call back handler registry from its configuration resource or file. This also has the side effect
-     * of configuring and registering the SASL client factory implementations using {@link DynamicSaslRegistrar}.
-     */
-    private CallbackHandlerRegistry()
+    static
     {
         // Register any configured SASL client factories.
         DynamicSaslRegistrar.registerSaslProviders();
@@ -120,12 +89,12 @@ public class CallbackHandlerRegistry
             FileUtils.openFileOrDefaultResource(filename, DEFAULT_RESOURCE_NAME,
                 CallbackHandlerRegistry.class.getClassLoader());
 
+        final Properties props = new Properties();
+
         try
         {
-            Properties props = new Properties();
+
             props.load(is);
-            parseProperties(props);
-            _logger.info("Callback handlers available for SASL mechanisms: " + _mechanisms);
         }
         catch (IOException e)
         {
@@ -146,32 +115,68 @@ public class CallbackHandlerRegistry
                 }
             }
         }
+
+        _instance = new CallbackHandlerRegistry(props);
+        _logger.info("Callback handlers available for SASL mechanisms: " + _instance._mechanisms);
+
     }
 
-    /*private InputStream openPropertiesInputStream(String filename)
+    /**
+     * Gets the singleton instance of this registry.
+     *
+     * @return The singleton instance of this registry.
+     */
+    public static CallbackHandlerRegistry getInstance()
+    {
+        return _instance;
+    }
+
+    public AMQCallbackHandler createCallbackHandler(final String mechanism)
     {
-        boolean useDefault = true;
-        InputStream is = null;
-        if (filename != null)
+        final Class<AMQCallbackHandler> mechanismClass = _mechanismToHandlerClassMap.get(mechanism);
+
+        if (mechanismClass == null)
         {
-            try
-            {
-                is = new BufferedInputStream(new FileInputStream(new File(filename)));
-                useDefault = false;
-            }
-            catch (FileNotFoundException e)
-            {
-                _logger.error("Unable to read from file " + filename + ": " + e, e);
-            }
+            throw new IllegalArgumentException("Mechanism " + mechanism + " not known");
         }
 
-        if (useDefault)
+        try
+        {
+            return mechanismClass.newInstance();
+        }
+        catch (InstantiationException e)
+        {
+            throw new IllegalArgumentException("Unable to create an instance of mechanism " + mechanism, e);
+        }
+        catch (IllegalAccessException e)
         {
-            is = CallbackHandlerRegistry.class.getResourceAsStream(DEFAULT_RESOURCE_NAME);
+            throw new IllegalArgumentException("Unable to create an instance of mechanism " + mechanism, e);
         }
+    }
 
-        return is;
-    }*/
+    /**
+     * Gets collections of supported SASL mechanism names, ordered by preference
+     *
+     * @return collection of SASL mechanism names.
+     */
+    public Collection<String> getMechanisms()
+    {
+        return Collections.unmodifiableCollection(_mechanisms);
+    }
+
+    /**
+     * Creates the call back handler registry from its configuration resource or file.
+     *
+     * This also has the side effect of configuring and registering the SASL client factory
+     * implementations using {@link DynamicSaslRegistrar}.
+     *
+     * This constructor is default protection to allow for effective unit testing.  Clients must use
+     * {@link #getInstance()} to obtain the singleton instance.
+     */
+    CallbackHandlerRegistry(final Properties props)
+    {
+        parseProperties(props);
+    }
 
     /**
      * Scans the specified properties as a mapping from IANA registered SASL mechanism to call back handler
@@ -183,20 +188,20 @@ public class CallbackHandlerRegistry
      */
     private void parseProperties(Properties props)
     {
+
+        final Map<Integer, String> mechanisms = new TreeMap<Integer, String>();
+
         Enumeration e = props.propertyNames();
         while (e.hasMoreElements())
         {
-            String propertyName = (String) e.nextElement();
-            int period = propertyName.indexOf(".");
-            if (period < 0)
-            {
-                _logger.warn("Unable to parse property " + propertyName + " when configuring SASL providers");
+            final String propertyName = (String) e.nextElement();
+            final String[] parts = propertyName.split("\\.", 2);
 
-                continue;
-            }
+            checkPropertyNameFormat(propertyName, parts);
 
-            String mechanism = propertyName.substring(period + 1);
-            String className = props.getProperty(propertyName);
+            final String mechanism = parts[0];
+            final int ordinal = getPropertyOrdinal(propertyName, parts);
+            final String className = props.getProperty(propertyName);
             Class clazz = null;
             try
             {
@@ -205,20 +210,11 @@ public class CallbackHandlerRegistry
                 {
                     _logger.warn("SASL provider " + clazz + " does not implement " + AMQCallbackHandler.class
                         + ". Skipping");
-
                     continue;
                 }
-
                 _mechanismToHandlerClassMap.put(mechanism, clazz);
-                if (_mechanisms == null)
-                {
-                    _mechanisms = mechanism;
-                }
-                else
-                {
-                    // one time cost
-                    _mechanisms = _mechanisms + " " + mechanism;
-                }
+
+                mechanisms.put(ordinal, mechanism);
             }
             catch (ClassNotFoundException ex)
             {
@@ -227,5 +223,91 @@ public class CallbackHandlerRegistry
                 continue;
             }
         }
+
+        _mechanisms = mechanisms.values();  // order guaranteed by keys of treemap (i.e. our ordinals)
+
+
+    }
+
+    private void checkPropertyNameFormat(final String propertyName, final String[] parts)
+    {
+        if (parts.length != 2)
+        {
+            throw new IllegalArgumentException("Unable to parse property " + propertyName + " when configuring SASL providers");
+        }
+    }
+
+    private int getPropertyOrdinal(final String propertyName, final String[] parts)
+    {
+        try
+        {
+            return Integer.parseInt(parts[1]);
+        }
+        catch(NumberFormatException nfe)
+        {
+            throw new IllegalArgumentException("Unable to parse property " + propertyName + " when configuring SASL providers", nfe);
+        }
+    }
+
+    /**
+     * Selects a SASL mechanism that is mutually available to both parties.  If more than one
+     * mechanism is mutually available the one appearing first (by ordinal) will be returned.
+     *
+     * @param peerMechanismList space separated list of mechanisms
+     * @return selected mechanism, or null if none available
+     */
+    public String selectMechanism(final String peerMechanismList)
+    {
+        final Set<String> peerList = mechListToSet(peerMechanismList);
+
+        return selectMechInternal(peerList, Collections.<String>emptySet());
+    }
+
+    /**
+     * Selects a SASL mechanism that is mutually available to both parties.
+     *
+     * @param peerMechanismList space separated list of mechanisms
+     * @param restrictionList space separated list of mechanisms
+     * @return selected mechanism, or null if none available
+     */
+    public String selectMechanism(final String peerMechanismList, final String restrictionList)
+    {
+        final Set<String> peerList = mechListToSet(peerMechanismList);
+        final Set<String> restrictionSet = mechListToSet(restrictionList);
+
+        return selectMechInternal(peerList, restrictionSet);
+    }
+
+    private String selectMechInternal(final Set<String> peerSet, final Set<String> restrictionSet)
+    {
+        for (final String mech : _mechanisms)
+        {
+            if (peerSet.contains(mech))
+            {
+                if (restrictionSet.isEmpty() || restrictionSet.contains(mech))
+                {
+                    return mech;
+                }
+            }
+        }
+
+        return null;
+    }
+
+    private Set<String> mechListToSet(final String mechanismList)
+    {
+        if (mechanismList == null)
+        {
+            return Collections.emptySet();
+        }
+
+        final StringTokenizer tokenizer = new StringTokenizer(mechanismList, " ");
+        final Set<String> mechanismSet = new HashSet<String>(tokenizer.countTokens());
+        while (tokenizer.hasMoreTokens())
+        {
+            mechanismSet.add(tokenizer.nextToken());
+        }
+        return Collections.unmodifiableSet(mechanismSet);
     }
+
 }
diff --git a/java/client/src/main/java/org/apache/qpid/client/security/CallbackHandlerRegistry.properties b/java/client/src/main/java/org/apache/qpid/client/security/CallbackHandlerRegistry.properties
index 1fcfde3579..b04a756e80 100644
--- a/java/client/src/main/java/org/apache/qpid/client/security/CallbackHandlerRegistry.properties
+++ b/java/client/src/main/java/org/apache/qpid/client/security/CallbackHandlerRegistry.properties
@@ -16,7 +16,17 @@
 # specific language governing permissions and limitations
 # under the License.
 #
-CallbackHandler.CRAM-MD5-HASHED=org.apache.qpid.client.security.UsernameHashedPasswordCallbackHandler
-CallbackHandler.CRAM-MD5=org.apache.qpid.client.security.UsernamePasswordCallbackHandler
-CallbackHandler.AMQPLAIN=org.apache.qpid.client.security.UsernamePasswordCallbackHandler
-CallbackHandler.PLAIN=org.apache.qpid.client.security.UsernamePasswordCallbackHandler
+
+#
+# Format:
+# <mechanism name>.ordinal=<implementation>
+#
+# @see CallbackHandlerRegistry
+#
+
+EXTERNAL.1=org.apache.qpid.client.security.UsernamePasswordCallbackHandler
+GSSAPI.2=org.apache.qpid.client.security.UsernamePasswordCallbackHandler
+CRAM-MD5-HASHED.3=org.apache.qpid.client.security.UsernameHashedPasswordCallbackHandler
+CRAM-MD5.4=org.apache.qpid.client.security.UsernamePasswordCallbackHandler
+AMQPLAIN.5=org.apache.qpid.client.security.UsernamePasswordCallbackHandler
+PLAIN.6=org.apache.qpid.client.security.UsernamePasswordCallbackHandler
diff --git a/java/client/src/main/java/org/apache/qpid/client/transport/ClientConnectionDelegate.java b/java/client/src/main/java/org/apache/qpid/client/transport/ClientConnectionDelegate.java
new file mode 100644
index 0000000000..1b483f6948
--- /dev/null
+++ b/java/client/src/main/java/org/apache/qpid/client/transport/ClientConnectionDelegate.java
@@ -0,0 +1,168 @@
+/*
+ *
+ * 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.qpid.client.transport;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import javax.security.sasl.Sasl;
+import javax.security.sasl.SaslClient;
+import javax.security.sasl.SaslException;
+
+import org.apache.qpid.client.security.AMQCallbackHandler;
+import org.apache.qpid.client.security.CallbackHandlerRegistry;
+import org.apache.qpid.jms.ConnectionURL;
+import org.apache.qpid.transport.ClientDelegate;
+import org.apache.qpid.transport.Connection;
+import org.apache.qpid.transport.ConnectionException;
+import org.apache.qpid.transport.ConnectionOpenOk;
+import org.apache.qpid.transport.ConnectionSettings;
+import org.apache.qpid.transport.util.Logger;
+import org.apache.qpid.util.Strings;
+import org.ietf.jgss.GSSContext;
+import org.ietf.jgss.GSSException;
+import org.ietf.jgss.GSSManager;
+import org.ietf.jgss.GSSName;
+import org.ietf.jgss.Oid;
+
+/**
+ *
+ */
+public class ClientConnectionDelegate extends ClientDelegate
+{
+    private static final Logger LOGGER = Logger.get(ClientDelegate.class);
+
+    private static final String KRB5_OID_STR = "1.2.840.113554.1.2.2";
+    protected static final Oid KRB5_OID;
+
+    static
+    {
+        Oid oid;
+        try
+        {
+            oid = new Oid(KRB5_OID_STR);
+        }
+        catch (GSSException ignore)
+        {
+            oid = null;
+        }
+
+        KRB5_OID = oid;
+    }
+
+    private final ConnectionURL _connectionURL;
+
+    /**
+     * @param settings
+     * @param connectionURL
+     */
+    public ClientConnectionDelegate(ConnectionSettings settings, ConnectionURL connectionURL)
+    {
+        super(settings);
+        this._connectionURL = connectionURL;
+    }
+
+    @Override
+    protected SaslClient createSaslClient(List<Object> brokerMechs) throws ConnectionException, SaslException
+    {
+        final String brokerMechanisms = Strings.join(" ", brokerMechs);
+        final String restrictionList = _conSettings.getSaslMechs();
+        final String selectedMech = CallbackHandlerRegistry.getInstance().selectMechanism(brokerMechanisms, restrictionList);
+        if (selectedMech == null)
+        {
+            throw new ConnectionException("Client and broker have no SASL mechanisms in common." +
+                    " Broker allows : " + brokerMechanisms +
+                    " Client has : " + CallbackHandlerRegistry.getInstance().getMechanisms()  +
+                    " Client restricted itself to : " + (restrictionList != null ? restrictionList : "no restriction"));
+        }
+
+        Map<String,Object> saslProps = new HashMap<String,Object>();
+        if (_conSettings.isUseSASLEncryption())
+        {
+            saslProps.put(Sasl.QOP, "auth-conf");
+        }
+
+        final AMQCallbackHandler handler = CallbackHandlerRegistry.getInstance().createCallbackHandler(selectedMech);
+        handler.initialise(_connectionURL);
+        final SaslClient sc = Sasl.createSaslClient(new String[] {selectedMech}, null, _conSettings.getSaslProtocol(), _conSettings.getSaslServerName(), saslProps, handler);
+
+        return sc;
+    }
+
+    @Override
+    public void connectionOpenOk(Connection conn, ConnectionOpenOk ok)
+    {
+        SaslClient sc = conn.getSaslClient();
+        if (sc != null)
+        {
+            if (sc.getMechanismName().equals("GSSAPI"))
+            {
+                String id = getKerberosUser();
+                if (id != null)
+                {
+                    conn.setUserID(id);
+                }
+            }
+            else if (sc.getMechanismName().equals("EXTERNAL"))
+            {
+                if (conn.getSecurityLayer() != null)
+                {
+                    conn.setUserID(conn.getSecurityLayer().getUserID());
+                }
+            }
+        }
+
+        super.connectionOpenOk(conn, ok);
+    }
+
+    private String getKerberosUser()
+    {
+        LOGGER.debug("Obtaining userID from kerberos");
+        String service = _conSettings.getSaslProtocol() + "@" + _conSettings.getSaslServerName();
+        GSSManager manager = GSSManager.getInstance();
+
+        try
+        {
+            GSSName acceptorName = manager.createName(service,
+                GSSName.NT_HOSTBASED_SERVICE, KRB5_OID);
+
+            GSSContext secCtx = manager.createContext(acceptorName,
+                                                      KRB5_OID,
+                                                      null,
+                                                      GSSContext.INDEFINITE_LIFETIME);
+
+            secCtx.initSecContext(new byte[0], 0, 1);
+
+            if (secCtx.getSrcName() != null)
+            {
+                return secCtx.getSrcName().toString();
+            }
+
+        }
+        catch (GSSException e)
+        {
+            LOGGER.warn("Unable to retrieve userID from Kerberos due to error",e);
+        }
+
+        return null;
+    }
+}
diff --git a/java/client/src/test/java/org/apache/qpid/client/security/CallbackHandlerRegistryTest.java b/java/client/src/test/java/org/apache/qpid/client/security/CallbackHandlerRegistryTest.java
new file mode 100644
index 0000000000..cc5d48fbef
--- /dev/null
+++ b/java/client/src/test/java/org/apache/qpid/client/security/CallbackHandlerRegistryTest.java
@@ -0,0 +1,185 @@
+/*
+ *
+ * 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.qpid.client.security;
+
+import java.io.IOException;
+import java.util.Properties;
+
+import javax.security.auth.callback.Callback;
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.auth.callback.UnsupportedCallbackException;
+
+import org.apache.qpid.jms.ConnectionURL;
+import org.apache.qpid.test.utils.QpidTestCase;
+
+
+/**
+ * Tests the ability of {@link CallbackHandlerRegistry} to correctly parse
+ * the properties describing the available callback handlers.   Ensures also
+ * that it is able to select the mechanism and create an implementation
+ * given a variety of starting conditions.
+ *
+ */
+public class CallbackHandlerRegistryTest extends QpidTestCase
+{
+    private CallbackHandlerRegistry _registry;  // Object under test
+
+    public void testCreateHandlerSuccess()
+    {
+        final Properties props = new Properties();
+        props.put("TESTA.1", TestACallbackHandler.class.getName());
+
+        _registry = new CallbackHandlerRegistry(props);
+        assertEquals(1,_registry.getMechanisms().size());
+
+        final CallbackHandler handler = _registry.createCallbackHandler("TESTA");
+        assertTrue(handler instanceof TestACallbackHandler);
+    }
+
+    public void testCreateHandlerForUnknownMechanismName()
+    {
+        final Properties props = new Properties();
+        props.put("TEST1.1", TestACallbackHandler.class.getName());
+
+        _registry = new CallbackHandlerRegistry(props);
+
+        try
+        {
+            _registry.createCallbackHandler("NOTFOUND");
+            fail("Exception not thrown");
+        }
+        catch (IllegalArgumentException iae)
+        {
+            // PASS
+        }
+    }
+
+    public void testSelectMechanism()
+    {
+        final Properties props = new Properties();
+        props.put("TESTA.1", TestACallbackHandler.class.getName());
+        props.put("TESTB.2", TestBCallbackHandler.class.getName());
+
+        _registry = new CallbackHandlerRegistry(props);
+        assertEquals(2,_registry.getMechanisms().size());
+
+        final String selectedMechanism = _registry.selectMechanism("TESTA");
+        assertEquals("TESTA", selectedMechanism);
+    }
+
+    public void testSelectReturnsFirstMutallyAvailableMechanism()
+    {
+        final Properties props = new Properties();
+        props.put("TESTA.1", TestACallbackHandler.class.getName());
+        props.put("TESTB.2", TestBCallbackHandler.class.getName());
+
+        _registry = new CallbackHandlerRegistry(props);
+
+        final String selectedMechanism = _registry.selectMechanism("TESTD TESTB TESTA");
+        // TESTA should be returned as it is higher than TESTB in the properties file.
+        assertEquals("Selected mechanism should respect the ordinal", "TESTA", selectedMechanism);
+    }
+
+    public void testRestrictedSelectReturnsMechanismFromRestrictedList()
+    {
+        final Properties props = new Properties();
+        props.put("TESTA.1", TestACallbackHandler.class.getName());
+        props.put("TESTB.2", TestBCallbackHandler.class.getName());
+        props.put("TESTC.3", TestCCallbackHandler.class.getName());
+
+        _registry = new CallbackHandlerRegistry(props);
+
+        final String selectedMechanism = _registry.selectMechanism("TESTC TESTB TESTA", "TESTB TESTC");
+        // TESTB should be returned as client has restricted the mechanism list to TESTB and TESTC
+        assertEquals("Selected mechanism should respect the ordinal and be limitted by restricted list","TESTB", selectedMechanism);
+    }
+
+    public void testOldPropertyFormatRejected()
+    {
+        final Properties props = new Properties();
+        props.put("CallbackHandler.TESTA", TestACallbackHandler.class.getName());
+
+        try
+        {
+            new CallbackHandlerRegistry(props);
+            fail("exception not thrown");
+        }
+        catch(IllegalArgumentException iae)
+        {
+            // PASS
+        }
+    }
+
+    public void testPropertyWithNonnumericalOrdinal()
+    {
+        final Properties props = new Properties();
+        props.put("TESTA.z", TestACallbackHandler.class.getName());
+        try
+        {
+            new CallbackHandlerRegistry(props);
+            fail("exception not thrown");
+        }
+        catch(IllegalArgumentException iae)
+        {
+            // PASS
+        }
+    }
+
+    public void testUnexpectedCallbackImplementationsIgnored()
+    {
+        final Properties props = new Properties();
+        props.put("TESTA.1", TestACallbackHandler.class.getName());
+        props.put("TESTB.2", "NotFound");
+        props.put("TESTC.3", "java.lang.String");
+
+        _registry = new CallbackHandlerRegistry(props);
+
+        assertEquals(1,_registry.getMechanisms().size());
+    }
+
+    static class TestACallbackHandler extends TestCallbackHandler
+    {
+    }
+
+    static class TestBCallbackHandler extends TestCallbackHandler
+    {
+    }
+
+    static class TestCCallbackHandler extends TestCallbackHandler
+    {
+    }
+
+    static abstract class TestCallbackHandler implements AMQCallbackHandler
+    {
+        @Override
+        public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException
+        {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public void initialise(ConnectionURL connectionURL)
+        {
+            throw new UnsupportedOperationException();
+        }
+    }
+
+}