QPID-4042 : [Java Broker] Add SSL Client Auth support

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

37 files changed, 832 insertions, 391 deletions

diff --git a/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ConnectionEndpoint.java b/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ConnectionEndpoint.java
index 4ad8b4196e..70e990d92e 100644
--- a/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ConnectionEndpoint.java
+++ b/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/ConnectionEndpoint.java
@@ -106,15 +106,15 @@ public class ConnectionEndpoint implements DescribedTypeConstructorRegistry.Sour
     private UnsignedInteger _desiredMaxFrameSize = UnsignedInteger.valueOf(DEFAULT_MAX_FRAME);

     private Runnable _onSaslCompleteTask;

 

-    private CallbackHandlerSource _callbackHandlersource;

+    private SaslServerProvider _saslServerProvider;

     private SaslServer _saslServer;

     private boolean _authenticated;

     private String _remoteHostname;

 

-    public ConnectionEndpoint(Container container, CallbackHandlerSource cbs)

+    public ConnectionEndpoint(Container container, SaslServerProvider cbs)

     {

         _container = container;

-        _callbackHandlersource = cbs;

+        _saslServerProvider = cbs;

         _requiresSASLClient = false;

         _requiresSASLServer = cbs != null;

     }

@@ -700,11 +700,7 @@ public class ConnectionEndpoint implements DescribedTypeConstructorRegistry.Sour
 

         try

         {

-            _saslServer = Sasl.createSaslServer(mechanism.toString(),

-                                                "AMQP",

-                                                "localhost",

-                                                null,

-                                                _callbackHandlersource.getHandler(mechanism.toString()));

+            _saslServer = _saslServerProvider.getSaslServer(mechanism.toString(), "localhost");

 

             // Process response from the client

             byte[] challenge = _saslServer.evaluateResponse(response != null ? response : new byte[0]);

diff --git a/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/CallbackHandlerSource.java b/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/SaslServerProvider.java
index 604279a21f..1b08488673 100644
--- a/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/CallbackHandlerSource.java
+++ b/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/transport/SaslServerProvider.java
@@ -20,9 +20,10 @@
 package org.apache.qpid.amqp_1_0.transport;
 
 
-import javax.security.auth.callback.CallbackHandler;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
 
-public interface CallbackHandlerSource
+public interface SaslServerProvider
 {
-    CallbackHandler getHandler(String mechanism);
+    SaslServer getSaslServer(String mechanism, String fqdn) throws SaslException;
 }
diff --git a/qpid/java/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/plugins/RuleSetTest.java b/qpid/java/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/plugins/RuleSetTest.java
index 4d46a32f45..f7cc60543d 100644
--- a/qpid/java/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/plugins/RuleSetTest.java
+++ b/qpid/java/broker-plugins/access-control/src/test/java/org/apache/qpid/server/security/access/plugins/RuleSetTest.java
@@ -40,12 +40,12 @@ import org.apache.qpid.test.utils.QpidTestCase;
 
 /**
  * This test checks that the {@link RuleSet} object which forms the core of the access control plugin performs correctly.
- * 
+ *
  * The ruleset is configured directly rather than using an external file by adding rules individually, calling the
  * {@link RuleSet#grant(Integer, String, Permission, Operation, ObjectType, ObjectProperties)} method. Then, the
  * access control mechanism is validated by checking whether operations would be authorised by calling the
  * {@link RuleSet#check(Principal, Operation, ObjectType, ObjectProperties)} method.
- * 
+ *
  * It ensure that permissions can be granted correctly on users directly, ACL groups (that is those
  * groups declared directly in the ACL itself), and External groups (that is a group from an External
  * Authentication Provider, such as an LDAP).
@@ -82,11 +82,11 @@ public class RuleSetTest extends QpidTestCase
     {
         assertDenyGrantAllow(subject, operation, objectType, ObjectProperties.EMPTY);
     }
-    
+
     public void assertDenyGrantAllow(Subject subject, Operation operation, ObjectType objectType, ObjectProperties properties)
     {
-        final Principal identity = UsernamePrincipal.getUsernamePrincipalFromSubject(subject);
-        
+        final Principal identity = subject.getPrincipals().iterator().next();
+
         assertEquals(Result.DENIED, _ruleSet.check(subject, operation, objectType, properties));
         _ruleSet.grant(0, identity.getName(), Permission.ALLOW, operation, objectType, properties);
         assertEquals(1, _ruleSet.getRuleCount());
@@ -99,7 +99,7 @@ public class RuleSetTest extends QpidTestCase
         assertEquals(_ruleSet.getRuleCount(), 0);
         assertEquals(_ruleSet.getDefault(), _ruleSet.check(_testSubject, Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY));
     }
-    
+
     public void testVirtualHostAccess() throws Exception
     {
         assertDenyGrantAllow(_testSubject, Operation.ACCESS, ObjectType.VIRTUALHOST);
@@ -114,7 +114,7 @@ public class RuleSetTest extends QpidTestCase
     {
         ObjectProperties properties = new ObjectProperties(_queueName);
         properties.put(ObjectProperties.Property.ROUTING_KEY, (String) null);
-        
+
         assertDenyGrantAllow(_testSubject, Operation.CREATE, ObjectType.QUEUE, properties);
     }
 
@@ -122,7 +122,7 @@ public class RuleSetTest extends QpidTestCase
     {
         ObjectProperties properties = new ObjectProperties(_exchangeName);
         properties.put(ObjectProperties.Property.TYPE, _exchangeType.asString());
-        
+
         assertDenyGrantAllow(_testSubject, Operation.CREATE, ObjectType.EXCHANGE, properties);
     }
 
@@ -144,15 +144,15 @@ public class RuleSetTest extends QpidTestCase
     {
         ObjectProperties temporary = new ObjectProperties();
         temporary.put(ObjectProperties.Property.AUTO_DELETE, Boolean.TRUE);
-        
+
         ObjectProperties normal = new ObjectProperties();
         normal.put(ObjectProperties.Property.AUTO_DELETE, Boolean.FALSE);
-        
+
         assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CONSUME, ObjectType.QUEUE, temporary));
         _ruleSet.grant(0, TEST_USER, Permission.ALLOW, Operation.CONSUME, ObjectType.QUEUE, temporary);
         assertEquals(1, _ruleSet.getRuleCount());
         assertEquals(Result.ALLOWED, _ruleSet.check(_testSubject, Operation.CONSUME, ObjectType.QUEUE, temporary));
-        
+
         // defer to global if exists, otherwise default answer - this is handled by the security manager
         assertEquals(Result.DEFER, _ruleSet.check(_testSubject, Operation.CONSUME, ObjectType.QUEUE, normal));
     }
@@ -164,17 +164,17 @@ public class RuleSetTest extends QpidTestCase
     {
         ObjectProperties temporary = new ObjectProperties(_queueName);
         temporary.put(ObjectProperties.Property.AUTO_DELETE, Boolean.TRUE);
-        
+
         ObjectProperties normal = new ObjectProperties(_queueName);
         normal.put(ObjectProperties.Property.AUTO_DELETE, Boolean.FALSE);
-        
+
         assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CONSUME, ObjectType.QUEUE, temporary));
 
         // should not matter if the temporary permission is processed first or last
         _ruleSet.grant(1, TEST_USER, Permission.ALLOW, Operation.CONSUME, ObjectType.QUEUE, normal);
         _ruleSet.grant(2, TEST_USER, Permission.ALLOW, Operation.CONSUME, ObjectType.QUEUE, temporary);
         assertEquals(2, _ruleSet.getRuleCount());
-        
+
         assertEquals(Result.ALLOWED, _ruleSet.check(_testSubject, Operation.CONSUME, ObjectType.QUEUE, normal));
         assertEquals(Result.ALLOWED, _ruleSet.check(_testSubject, Operation.CONSUME, ObjectType.QUEUE, temporary));
     }
@@ -186,17 +186,17 @@ public class RuleSetTest extends QpidTestCase
     {
         ObjectProperties temporary = new ObjectProperties(_queueName);
         temporary.put(ObjectProperties.Property.AUTO_DELETE, Boolean.TRUE);
-        
+
         ObjectProperties normal = new ObjectProperties(_queueName);
         normal.put(ObjectProperties.Property.AUTO_DELETE, Boolean.FALSE);
-        
+
         assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CONSUME, ObjectType.QUEUE, temporary));
 
         // should not matter if the temporary permission is processed first or last
         _ruleSet.grant(1, TEST_USER, Permission.ALLOW, Operation.CONSUME, ObjectType.QUEUE, temporary);
         _ruleSet.grant(2, TEST_USER, Permission.ALLOW, Operation.CONSUME, ObjectType.QUEUE, normal);
         assertEquals(2, _ruleSet.getRuleCount());
-        
+
         assertEquals(Result.ALLOWED, _ruleSet.check(_testSubject, Operation.CONSUME, ObjectType.QUEUE, normal));
         assertEquals(Result.ALLOWED, _ruleSet.check(_testSubject, Operation.CONSUME, ObjectType.QUEUE, temporary));
     }
@@ -204,7 +204,7 @@ public class RuleSetTest extends QpidTestCase
     /*
      * Test different rules for temporary queues.
      */
-    
+
     /**
      * The more generic rule first is used, so both requests are allowed.
      */
@@ -213,18 +213,18 @@ public class RuleSetTest extends QpidTestCase
         ObjectProperties named = new ObjectProperties(_queueName);
         ObjectProperties namedTemporary = new ObjectProperties(_queueName);
         namedTemporary.put(ObjectProperties.Property.AUTO_DELETE, Boolean.TRUE);
-        
+
         assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, named));
         assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, namedTemporary));
 
         _ruleSet.grant(1, TEST_USER, Permission.ALLOW, Operation.CREATE, ObjectType.QUEUE, named);
         _ruleSet.grant(2, TEST_USER, Permission.DENY, Operation.CREATE, ObjectType.QUEUE, namedTemporary);
         assertEquals(2, _ruleSet.getRuleCount());
-        
+
         assertEquals(Result.ALLOWED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, named));
         assertEquals(Result.ALLOWED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, namedTemporary));
     }
-    
+
     /**
      * The more specific rule is first, so those requests are denied.
      */
@@ -233,18 +233,18 @@ public class RuleSetTest extends QpidTestCase
         ObjectProperties named = new ObjectProperties(_queueName);
         ObjectProperties namedTemporary = new ObjectProperties(_queueName);
         namedTemporary.put(ObjectProperties.Property.AUTO_DELETE, Boolean.TRUE);
-        
+
         assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, named));
         assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, namedTemporary));
 
         _ruleSet.grant(1, TEST_USER, Permission.DENY, Operation.CREATE, ObjectType.QUEUE, namedTemporary);
         _ruleSet.grant(2, TEST_USER, Permission.ALLOW, Operation.CREATE, ObjectType.QUEUE, named);
         assertEquals(2, _ruleSet.getRuleCount());
-        
+
         assertEquals(Result.ALLOWED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, named));
         assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, namedTemporary));
     }
-    
+
     /**
      * The more specific rules are first, so those requests are denied.
      */
@@ -255,7 +255,7 @@ public class RuleSetTest extends QpidTestCase
         namedTemporary.put(ObjectProperties.Property.AUTO_DELETE, Boolean.TRUE);
         ObjectProperties namedDurable = new ObjectProperties(_queueName);
         namedDurable.put(ObjectProperties.Property.DURABLE, Boolean.TRUE);
-        
+
         assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, named));
         assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, namedTemporary));
         assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, namedDurable));
@@ -264,48 +264,48 @@ public class RuleSetTest extends QpidTestCase
         _ruleSet.grant(2, TEST_USER, Permission.DENY, Operation.CREATE, ObjectType.QUEUE, namedDurable);
         _ruleSet.grant(3, TEST_USER, Permission.ALLOW, Operation.CREATE, ObjectType.QUEUE, named);
         assertEquals(3, _ruleSet.getRuleCount());
-        
+
         assertEquals(Result.ALLOWED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, named));
         assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, namedTemporary));
         assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, namedDurable));
     }
-    
+
     public void testNamedTemporaryQueueAllowed()
     {
         ObjectProperties named = new ObjectProperties(_queueName);
         ObjectProperties namedTemporary = new ObjectProperties(_queueName);
         namedTemporary.put(ObjectProperties.Property.AUTO_DELETE, Boolean.TRUE);
-        
+
         assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, named));
         assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, namedTemporary));
 
         _ruleSet.grant(1, TEST_USER, Permission.ALLOW, Operation.CREATE, ObjectType.QUEUE, namedTemporary);
         _ruleSet.grant(2, TEST_USER, Permission.ALLOW, Operation.CREATE, ObjectType.QUEUE, named);
         assertEquals(2, _ruleSet.getRuleCount());
-        
+
         assertEquals(Result.ALLOWED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, named));
         assertEquals(Result.ALLOWED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, namedTemporary));
     }
-    
+
     public void testNamedTemporaryQueueDeniedAllowed()
     {
         ObjectProperties named = new ObjectProperties(_queueName);
         ObjectProperties namedTemporary = new ObjectProperties(_queueName);
         namedTemporary.put(ObjectProperties.Property.AUTO_DELETE, Boolean.TRUE);
-        
+
         assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, named));
         assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, namedTemporary));
 
         _ruleSet.grant(1, TEST_USER, Permission.ALLOW, Operation.CREATE, ObjectType.QUEUE, namedTemporary);
         _ruleSet.grant(2, TEST_USER, Permission.DENY, Operation.CREATE, ObjectType.QUEUE, named);
         assertEquals(2, _ruleSet.getRuleCount());
-        
+
         assertEquals(Result.DENIED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, named));
         assertEquals(Result.ALLOWED, _ruleSet.check(_testSubject, Operation.CREATE, ObjectType.QUEUE, namedTemporary));
     }
 
-    /** 
-     * Tests support for the {@link Rule#ALL} keyword. 
+    /**
+     * Tests support for the {@link Rule#ALL} keyword.
      */
     public void testAllowToAll()
     {
@@ -316,13 +316,13 @@ public class RuleSetTest extends QpidTestCase
         assertEquals(Result.ALLOWED, _ruleSet.check(TestPrincipalUtils.createTestSubject("userb"),Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY));
     }
 
-    /** 
+    /**
      * Tests support for ACL groups (i.e. inline groups declared in the ACL file itself).
      */
     public void testAclGroupsSupported()
     {
-        assertTrue(_ruleSet.addGroup("aclgroup", Arrays.asList(new String[] {"usera", "userb"})));  
-        
+        assertTrue(_ruleSet.addGroup("aclgroup", Arrays.asList(new String[] {"usera", "userb"})));
+
         _ruleSet.grant(1, "aclgroup", Permission.ALLOW, Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY);
         assertEquals(1, _ruleSet.getRuleCount());
 
@@ -331,14 +331,14 @@ public class RuleSetTest extends QpidTestCase
         assertEquals(Result.DEFER, _ruleSet.check(TestPrincipalUtils.createTestSubject("userc"),Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY));
     }
 
-    /** 
+    /**
      * Tests support for nested ACL groups.
      */
     public void testNestedAclGroupsSupported()
     {
         assertTrue(_ruleSet.addGroup("aclgroup1", Arrays.asList(new String[] {"userb"})));
-        assertTrue(_ruleSet.addGroup("aclgroup2", Arrays.asList(new String[] {"usera", "aclgroup1"})));  
-        
+        assertTrue(_ruleSet.addGroup("aclgroup2", Arrays.asList(new String[] {"usera", "aclgroup1"})));
+
         _ruleSet.grant(1, "aclgroup2", Permission.ALLOW, Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY);
         assertEquals(1, _ruleSet.getRuleCount());
 
@@ -346,7 +346,7 @@ public class RuleSetTest extends QpidTestCase
         assertEquals(Result.ALLOWED, _ruleSet.check(TestPrincipalUtils.createTestSubject("userb"),Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY));
     }
 
-    /** 
+    /**
      * Tests support for nested External groups (i.e. those groups coming from an external source such as an LDAP).
      */
     public void testExternalGroupsSupported()
@@ -358,7 +358,7 @@ public class RuleSetTest extends QpidTestCase
         assertEquals(Result.ALLOWED, _ruleSet.check(TestPrincipalUtils.createTestSubject("usera", "extgroup1"),Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY));
         assertEquals(Result.DENIED, _ruleSet.check(TestPrincipalUtils.createTestSubject("userb", "extgroup2"),Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY));
     }
-    
+
     /**
      * Rule order in the ACL determines the outcome of the check.  This test ensures that a user who is
      * granted explicit permission on an object, is granted that access even although late a group
@@ -367,7 +367,7 @@ public class RuleSetTest extends QpidTestCase
     public void testAllowDeterminedByRuleOrder()
     {
         assertTrue(_ruleSet.addGroup("aclgroup", Arrays.asList(new String[] {"usera"})));
-        
+
         _ruleSet.grant(1, "usera", Permission.ALLOW, Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY);
         _ruleSet.grant(2, "aclgroup", Permission.DENY, Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY);
         assertEquals(2, _ruleSet.getRuleCount());
@@ -382,10 +382,10 @@ public class RuleSetTest extends QpidTestCase
     public void testDenyDeterminedByRuleOrder()
     {
         assertTrue(_ruleSet.addGroup("aclgroup", Arrays.asList(new String[] {"usera"})));
-        
+
         _ruleSet.grant(1, "aclgroup", Permission.DENY, Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY);
         _ruleSet.grant(2, "usera", Permission.ALLOW, Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY);
-        
+
         assertEquals(2, _ruleSet.getRuleCount());
 
         assertEquals(Result.DENIED, _ruleSet.check(TestPrincipalUtils.createTestSubject("usera"),Operation.ACCESS, ObjectType.VIRTUALHOST, ObjectProperties.EMPTY));
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/Broker.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/Broker.java
index 2b43d41c7a..90603777d1 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/Broker.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/Broker.java
@@ -20,10 +20,20 @@
  */
 package org.apache.qpid.server;
 
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.util.EnumSet;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+import javax.net.ssl.SSLContext;
 import org.apache.log4j.Logger;
 import org.apache.log4j.PropertyConfigurator;
 import org.apache.log4j.xml.QpidLog4JConfigurator;
-
 import org.apache.qpid.server.configuration.ServerConfiguration;
 import org.apache.qpid.server.configuration.ServerNetworkTransportConfiguration;
 import org.apache.qpid.server.configuration.management.ConfigurationManagementMBean;
@@ -46,18 +56,6 @@ import org.apache.qpid.transport.network.Transport;
 
 import static org.apache.qpid.transport.ConnectionSettings.WILDCARD_ADDRESS;
 
-import javax.net.ssl.SSLContext;
-import java.io.File;
-import java.io.IOException;
-import java.io.InputStream;
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
-import java.util.EnumSet;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Properties;
-import java.util.Set;
-
 public class Broker
 {
     private static final Logger LOGGER = Logger.getLogger(Broker.class);
@@ -271,7 +269,21 @@ public class Broker
                 final String keystorePassword = serverConfig.getConnectorKeyStorePassword();
                 final String keystoreType = serverConfig.getConnectorKeyStoreType();
                 final String keyManagerFactoryAlgorithm = serverConfig.getConnectorKeyManagerFactoryAlgorithm();
-                final SSLContext sslContext = SSLContextFactory.buildServerContext(keystorePath, keystorePassword, keystoreType, keyManagerFactoryAlgorithm);
+                final SSLContext sslContext;
+                if(serverConfig.getConnectorTrustStorePath()!=null)
+                {
+                    sslContext = SSLContextFactory.buildClientContext(serverConfig.getConnectorTrustStorePath(),
+                                                                      serverConfig.getConnectorTrustStorePassword(),
+                                                                      serverConfig.getConnectorTrustStoreType(),
+                                                                      serverConfig.getConnectorTrustManagerFactoryAlgorithm(),
+                                                                      keystorePath,
+                                                                      keystorePassword, keystoreType, keyManagerFactoryAlgorithm,
+                                                                      serverConfig.getCertAlias());
+                }
+                else
+                {
+                    sslContext = SSLContextFactory.buildServerContext(keystorePath, keystorePassword, keystoreType, keyManagerFactoryAlgorithm);
+                }
 
                 for(int sslPort : sslPorts)
                 {
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java
index 651fd26059..0e538a13a6 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java
@@ -20,6 +20,16 @@
 
 package org.apache.qpid.server.configuration;
 
+import java.io.File;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Map.Entry;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.TrustManagerFactory;
 import org.apache.commons.configuration.CompositeConfiguration;
 import org.apache.commons.configuration.Configuration;
 import org.apache.commons.configuration.ConfigurationException;
@@ -28,7 +38,6 @@ import org.apache.commons.configuration.HierarchicalConfiguration;
 import org.apache.commons.configuration.SystemConfiguration;
 import org.apache.commons.configuration.XMLConfiguration;
 import org.apache.log4j.Logger;
-
 import org.apache.qpid.server.configuration.plugins.ConfigurationPlugin;
 import org.apache.qpid.server.exchange.DefaultExchangeFactory;
 import org.apache.qpid.server.protocol.AmqpProtocolVersion;
@@ -40,17 +49,6 @@ import org.apache.qpid.server.virtualhost.VirtualHostRegistry;
 
 import static org.apache.qpid.transport.ConnectionSettings.WILDCARD_ADDRESS;
 
-import java.io.File;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Locale;
-import java.util.Map;
-import java.util.Map.Entry;
-
-import javax.net.ssl.KeyManagerFactory;
-
 public class ServerConfiguration extends ConfigurationPlugin
 {
     protected static final Logger _logger = Logger.getLogger(ServerConfiguration.class);
@@ -182,7 +180,7 @@ public class ServerConfiguration extends ConfigurationPlugin
      * This has been made a two step process to allow the Plugin Manager and
      * Configuration Manager to be initialised in the Application Registry.
      * <p>
-     * If using this ServerConfiguration via an ApplicationRegistry there is no 
+     * If using this ServerConfiguration via an ApplicationRegistry there is no
      * need to explicitly call {@link #initialise()} as this is done via the
      * {@link ApplicationRegistry#initialise()} method.
      *
@@ -204,12 +202,12 @@ public class ServerConfiguration extends ConfigurationPlugin
      * Called by {@link ApplicationRegistry#initialise()}.
      * <p>
      * NOTE: A DEFAULT ApplicationRegistry must exist when using this method
-     * or a new ApplicationRegistry will be created. 
+     * or a new ApplicationRegistry will be created.
      *
      * @throws ConfigurationException
      */
     public void initialise() throws ConfigurationException
-    {	
+    {
         setConfiguration("", getConfig());
         setupVirtualHosts(getConfig());
     }
@@ -224,10 +222,10 @@ public class ServerConfiguration extends ConfigurationPlugin
     {
         // Support for security.jmx.access was removed when JMX access rights were incorporated into the main ACL.
         // This ensure that users remove the element from their configuration file.
-        
+
         if (getListValue("security.jmx.access").size() > 0)
         {
-            String message = "Validation error : security/jmx/access is no longer a supported element within the configuration xml." 
+            String message = "Validation error : security/jmx/access is no longer a supported element within the configuration xml."
                     + (_configFile == null ? "" : " Configuration file : " + _configFile);
             throw new ConfigurationException(message);
         }
@@ -241,7 +239,7 @@ public class ServerConfiguration extends ConfigurationPlugin
 
         if (getListValue("security.principal-databases.principal-database(0).class").size() > 0)
         {
-            String message = "Validation error : security/principal-databases is no longer supported within the configuration xml." 
+            String message = "Validation error : security/principal-databases is no longer supported within the configuration xml."
                     + (_configFile == null ? "" : " Configuration file : " + _configFile);
             throw new ConfigurationException(message);
         }
@@ -475,7 +473,7 @@ public class ServerConfiguration extends ConfigurationPlugin
             Configuration newConfig = parseConfig(_configFile);
             setConfiguration("", newConfig);
             ApplicationRegistry.getInstance().getSecurityManager().configureHostPlugins(this);
-			
+
             // Reload virtualhosts from correct location
             Configuration newVhosts;
             if (_vhostsFile == null)
@@ -500,12 +498,12 @@ public class ServerConfiguration extends ConfigurationPlugin
             _logger.warn(SECURITY_CONFIG_RELOADED);
         }
     }
-    
+
     public String getQpidWork()
     {
         return System.getProperty(QPID_WORK, System.getProperty("java.io.tmpdir"));
     }
-    
+
     public String getQpidHome()
     {
         return System.getProperty(QPID_HOME);
@@ -550,12 +548,12 @@ public class ServerConfiguration extends ConfigurationPlugin
     {
         return _virtualHosts.keySet().toArray(new String[_virtualHosts.size()]);
     }
-    
+
     public String getPluginDirectory()
     {
         return getStringValue("plugin-directory");
     }
-    
+
     public String getCacheDirectory()
     {
         return getStringValue("cache-directory");
@@ -773,7 +771,7 @@ public class ServerConfiguration extends ConfigurationPlugin
     {
         return getBooleanValue("connector.ssl.sslOnly");
     }
-    
+
     public List getSSLPorts()
     {
         return getListValue("connector.ssl.port", Collections.<Integer>singletonList(DEFAULT_SSL_PORT));
@@ -804,6 +802,41 @@ public class ServerConfiguration extends ConfigurationPlugin
         return getStringValue("connector.ssl.keyManagerFactoryAlgorithm", fallback);
     }
 
+    public String getConnectorTrustStorePath()
+    {
+        return getStringValue("connector.ssl.trustStorePath", null);
+    }
+
+    public String getConnectorTrustStorePassword()
+    {
+        return getStringValue("connector.ssl.trustStorePassword", null);
+    }
+
+    public String getConnectorTrustStoreType()
+    {
+        return getStringValue("connector.ssl.trustStoreType", "JKS");
+    }
+
+    public String getConnectorTrustManagerFactoryAlgorithm()
+    {
+        return getStringValue("connector.ssl.trustManagerFactoryAlgorithm", TrustManagerFactory.getDefaultAlgorithm());
+    }
+
+    public String getCertAlias()
+    {
+        return getStringValue("connector.ssl.certAlias", null);
+    }
+
+    public boolean needClientAuth()
+    {
+        return getConfig().getBoolean("connector.ssl.needClientAuth", false);
+    }
+
+    public boolean wantClientAuth()
+    {
+        return getConfig().getBoolean("connector.ssl.wantClientAuth", false);
+    }
+
     public String getDefaultVirtualHost()
     {
         return getStringValue("virtualhosts.default");
@@ -812,7 +845,7 @@ public class ServerConfiguration extends ConfigurationPlugin
     public void setDefaultVirtualHost(String vhost)
     {
          getConfig().setProperty("virtualhosts.default", vhost);
-    }    
+    }
 
     public void setHousekeepingCheckPeriod(long value)
     {
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerNetworkTransportConfiguration.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerNetworkTransportConfiguration.java
index f6fe47b996..51dcc38c47 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerNetworkTransportConfiguration.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerNetworkTransportConfiguration.java
@@ -28,7 +28,7 @@ public class ServerNetworkTransportConfiguration implements NetworkTransportConf
     private final String _transport;
     private InetSocketAddress _address;
 
-    public ServerNetworkTransportConfiguration(final ServerConfiguration serverConfig, 
+    public ServerNetworkTransportConfiguration(final ServerConfiguration serverConfig,
                                                final InetSocketAddress address,
                                                final String transport)
     {
@@ -76,4 +76,15 @@ public class ServerNetworkTransportConfiguration implements NetworkTransportConf
     {
         return _address;
     }
+
+    public boolean needClientAuth()
+    {
+        return _serverConfig.needClientAuth();
+    }
+
+    @Override
+    public boolean wantClientAuth()
+    {
+        return _serverConfig.wantClientAuth();
+    }
 }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionSecureOkMethodHandler.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionSecureOkMethodHandler.java
index ac322c4e8c..b8c8411c5d 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionSecureOkMethodHandler.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionSecureOkMethodHandler.java
@@ -7,9 +7,9 @@
  * 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
@@ -92,7 +92,7 @@ public class ConnectionSecureOkMethodHandler implements StateAwareMethodListener
             case SUCCESS:
                 if (_logger.isInfoEnabled())
                 {
-                    _logger.info("Connected as: " + UsernamePrincipal.getUsernamePrincipalFromSubject(authResult.getSubject()));
+                    _logger.info("Connected as: " + authResult.getSubject());
                 }
                 stateManager.changeState(AMQState.CONNECTION_NOT_TUNED);
 
@@ -102,7 +102,7 @@ public class ConnectionSecureOkMethodHandler implements StateAwareMethodListener
                                                                 ApplicationRegistry.getInstance().getConfiguration().getHeartBeatDelay());
                 session.writeFrame(tuneBody.generateFrame(0));
                 session.setAuthorizedSubject(authResult.getSubject());
-                disposeSaslServer(session);                
+                disposeSaslServer(session);
                 break;
             case CONTINUE:
                 stateManager.changeState(AMQState.CONNECTION_NOT_AUTH);
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java
index d9979ed2dc..a522b9f60f 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/ConnectionStartOkMethodHandler.java
@@ -7,9 +7,9 @@
  * 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
@@ -61,15 +61,15 @@ public class ConnectionStartOkMethodHandler implements StateAwareMethodListener<
     public void methodReceived(AMQStateManager stateManager, ConnectionStartOkBody body, int channelId) throws AMQException
     {
         AMQProtocolSession session = stateManager.getProtocolSession();
-        
+
         _logger.info("SASL Mechanism selected: " + body.getMechanism());
         _logger.info("Locale selected: " + body.getLocale());
 
         AuthenticationManager authMgr = stateManager.getAuthenticationManager();
         SaslServer ss = null;
         try
-        {                       
-            ss = authMgr.createSaslServer(String.valueOf(body.getMechanism()), session.getLocalFQDN());
+        {
+            ss = authMgr.createSaslServer(String.valueOf(body.getMechanism()), session.getLocalFQDN(), session.getPeerPrincipal());
 
             if (ss == null)
             {
@@ -106,7 +106,7 @@ public class ConnectionStartOkMethodHandler implements StateAwareMethodListener<
                 case SUCCESS:
                     if (_logger.isInfoEnabled())
                     {
-                        _logger.info("Connected as: " + UsernamePrincipal.getUsernamePrincipalFromSubject(authResult.getSubject()));
+                        _logger.info("Connected as: " + authResult.getSubject());
                     }
                     session.setAuthorizedSubject(authResult.getSubject());
 
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/management/JMXManagedObjectRegistry.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/management/JMXManagedObjectRegistry.java
index 116f64a8bf..869a816cf1 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/management/JMXManagedObjectRegistry.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/management/JMXManagedObjectRegistry.java
@@ -20,17 +20,28 @@
  */
 package org.apache.qpid.server.management;
 
-import org.apache.commons.configuration.ConfigurationException;
-import org.apache.log4j.Logger;
-
-import org.apache.qpid.AMQException;
-import org.apache.qpid.server.logging.actors.CurrentActor;
-import org.apache.qpid.server.logging.messages.ManagementConsoleMessages;
-import org.apache.qpid.server.registry.ApplicationRegistry;
-import org.apache.qpid.server.registry.IApplicationRegistry;
-import org.apache.qpid.server.security.auth.rmi.RMIPasswordAuthenticator;
-import org.apache.qpid.server.security.auth.sasl.UsernamePrincipal;
-
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.lang.management.ManagementFactory;
+import java.lang.reflect.Proxy;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.UnknownHostException;
+import java.rmi.AlreadyBoundException;
+import java.rmi.NoSuchObjectException;
+import java.rmi.NotBoundException;
+import java.rmi.registry.LocateRegistry;
+import java.rmi.registry.Registry;
+import java.rmi.server.RMIClientSocketFactory;
+import java.rmi.server.RMIServerSocketFactory;
+import java.rmi.server.UnicastRemoteObject;
+import java.security.Principal;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 import javax.management.JMException;
 import javax.management.MBeanServer;
 import javax.management.MBeanServerFactory;
@@ -49,32 +60,23 @@ import javax.management.remote.rmi.RMIServerImpl;
 import javax.rmi.ssl.SslRMIClientSocketFactory;
 import javax.rmi.ssl.SslRMIServerSocketFactory;
 import javax.security.auth.Subject;
-import java.io.File;
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.lang.management.ManagementFactory;
-import java.lang.reflect.Proxy;
-import java.net.*;
-import java.rmi.AlreadyBoundException;
-import java.rmi.NoSuchObjectException;
-import java.rmi.NotBoundException;
-import java.rmi.registry.LocateRegistry;
-import java.rmi.registry.Registry;
-import java.rmi.server.RMIClientSocketFactory;
-import java.rmi.server.RMIServerSocketFactory;
-import java.rmi.server.UnicastRemoteObject;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.log4j.Logger;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.server.logging.actors.CurrentActor;
+import org.apache.qpid.server.logging.messages.ManagementConsoleMessages;
+import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.server.registry.IApplicationRegistry;
+import org.apache.qpid.server.security.auth.rmi.RMIPasswordAuthenticator;
 
 /**
- * This class starts up an MBeanserver. If out of the box agent has been enabled then there are no 
+ * This class starts up an MBeanserver. If out of the box agent has been enabled then there are no
  * security features implemented like user authentication and authorisation.
  */
 public class JMXManagedObjectRegistry implements ManagedObjectRegistry
 {
     private static final Logger _log = Logger.getLogger(JMXManagedObjectRegistry.class);
-    
+
     private final MBeanServer _mbeanServer;
     private JMXConnectorServer _cs;
     private Registry _rmiRegistry;
@@ -105,7 +107,7 @@ public class JMXManagedObjectRegistry implements ManagedObjectRegistry
     {
 
         CurrentActor.get().message(ManagementConsoleMessages.STARTUP());
-        
+
         //check if system properties are set to use the JVM's out-of-the-box JMXAgent
         if (areOutOfTheBoxJMXOptionsSet())
         {
@@ -158,10 +160,10 @@ public class JMXManagedObjectRegistry implements ManagedObjectRegistry
                 }
                 if (!ksf.canRead())
                 {
-                    throw new FileNotFoundException("Cannot read JMX management SSL keystore file: " 
+                    throw new FileNotFoundException("Cannot read JMX management SSL keystore file: "
                                                     + ksf +  ". Check permissions.");
                 }
-                
+
                 CurrentActor.get().message(ManagementConsoleMessages.SSL_KEYSTORE(ksf.getAbsolutePath()));
             }
 
@@ -199,9 +201,9 @@ public class JMXManagedObjectRegistry implements ManagedObjectRegistry
         env.put(JMXConnectorServer.AUTHENTICATOR, rmipa);
 
         /*
-         * Start a RMI registry on the management port, to hold the JMX RMI ConnectorServer stub. 
+         * Start a RMI registry on the management port, to hold the JMX RMI ConnectorServer stub.
          * Using custom socket factory to prevent anyone (including us unfortunately) binding to the registry using RMI.
-         * As a result, only binds made using the object reference will succeed, thus securing it from external change. 
+         * As a result, only binds made using the object reference will succeed, thus securing it from external change.
          */
         System.setProperty("java.rmi.server.randomIDs", "true");
         if(_useCustomSocketFactory)
@@ -212,17 +214,17 @@ public class JMXManagedObjectRegistry implements ManagedObjectRegistry
         {
             _rmiRegistry = LocateRegistry.createRegistry(_jmxPortRegistryServer, null, null);
         }
-        
+
         CurrentActor.get().message(ManagementConsoleMessages.LISTENING("RMI Registry", _jmxPortRegistryServer));
 
         /*
-         * We must now create the RMI ConnectorServer manually, as the JMX Factory methods use RMI calls 
+         * We must now create the RMI ConnectorServer manually, as the JMX Factory methods use RMI calls
          * to bind the ConnectorServer to the registry, which will now fail as for security we have
-         * locked it from any RMI based modifications, including our own. Instead, we will manually bind 
+         * locked it from any RMI based modifications, including our own. Instead, we will manually bind
          * the RMIConnectorServer stub to the registry using its object reference, which will still succeed.
-         * 
+         *
          * The registry is exported on the defined management port 'port'. We will export the RMIConnectorServer
-         * on 'port +1'. Use of these two well-defined ports will ease any navigation through firewall's. 
+         * on 'port +1'. Use of these two well-defined ports will ease any navigation through firewall's.
          */
         final Map<String, String> connectionIdUsernameMap = new ConcurrentHashMap<String, String>();
         final RMIServerImpl rmiConnectorServerStub = new RMIJRMPServerImpl(_jmxPortConnectorServer, csf, ssf, env)
@@ -240,8 +242,8 @@ public class JMXManagedObjectRegistry implements ManagedObjectRegistry
             protected RMIConnection makeClient(String connectionId, Subject subject) throws IOException
             {
                 final RMIConnection makeClient = super.makeClient(connectionId, subject);
-                final UsernamePrincipal usernamePrincipalFromSubject = UsernamePrincipal.getUsernamePrincipalFromSubject(subject);
-                connectionIdUsernameMap.put(connectionId, usernamePrincipalFromSubject.getName());
+                final Principal principal = subject.getPrincipals().iterator().next();
+                connectionIdUsernameMap.put(connectionId, principal.getName());
                 return makeClient;
             }
         };
@@ -273,32 +275,32 @@ public class JMXManagedObjectRegistry implements ManagedObjectRegistry
 
         final JMXServiceURL internalUrl = new JMXServiceURL("rmi", hostname, _jmxPortConnectorServer);
         _cs = new RMIConnectorServer(internalUrl, env, rmiConnectorServerStub, _mbeanServer)
-        {   
-            @Override  
+        {
+            @Override
             public synchronized void start() throws IOException
-            {   
+            {
                 try
-                {   
-                    //manually bind the connector server to the registry at key 'jmxrmi', like the out-of-the-box agent                        
+                {
+                    //manually bind the connector server to the registry at key 'jmxrmi', like the out-of-the-box agent
                     _rmiRegistry.bind("jmxrmi", rmiConnectorServerStub);
                 }
                 catch (AlreadyBoundException abe)
-                {   
+                {
                     //key was already in use. shouldnt happen here as its a new registry, unbindable by normal means.
 
                     //IOExceptions are the only checked type throwable by the method, wrap and rethrow
-                    IOException ioe = new IOException(abe.getMessage());   
-                    ioe.initCause(abe);   
-                    throw ioe;   
+                    IOException ioe = new IOException(abe.getMessage());
+                    ioe.initCause(abe);
+                    throw ioe;
                 }
 
                 //now do the normal tasks
-                super.start();   
+                super.start();
             }
 
-            @Override  
+            @Override
             public synchronized void stop() throws IOException
-            {   
+            {
                 try
                 {
                     if (_rmiRegistry != null)
@@ -310,20 +312,20 @@ public class JMXManagedObjectRegistry implements ManagedObjectRegistry
                 {
                     //ignore
                 }
-                
+
                 //now do the normal tasks
                 super.stop();
             }
-            
-            @Override  
+
+            @Override
             public JMXServiceURL getAddress()
             {
                 //must return our pre-crafted url that includes the full details, inc JNDI details
                 return externalUrl;
-            }   
+            }
+
+        };
 
-        };   
-        
 
         //Add the custom invoker as an MBeanServerForwarder, and start the RMIConnectorServer.
         MBeanServerForwarder mbsf = MBeanInvocationHandlerImpl.newProxyInstance();
@@ -359,14 +361,14 @@ public class JMXManagedObjectRegistry implements ManagedObjectRegistry
     }
 
     /*
-     * Custom RMIServerSocketFactory class, used to prevent updates to the RMI registry. 
+     * Custom RMIServerSocketFactory class, used to prevent updates to the RMI registry.
      * Supplied to the registry at creation, this will prevent RMI-based operations on the
      * registry such as attempting to bind a new object, thereby securing it from tampering.
      * This is accomplished by always returning null when attempting to determine the address
      * of the caller, thus ensuring the registry will refuse the attempt. Calls to bind etc
      * made using the object reference will not be affected and continue to operate normally.
      */
-    
+
     private static class CustomRMIServerSocketFactory implements RMIServerSocketFactory
     {
 
@@ -444,7 +446,7 @@ public class JMXManagedObjectRegistry implements ManagedObjectRegistry
                 _log.error("Exception while closing the JMX ConnectorServer: " + e.getMessage());
             }
         }
-        
+
         if (_rmiRegistry != null)
         {
             // Stopping the RMI registry
@@ -458,7 +460,7 @@ public class JMXManagedObjectRegistry implements ManagedObjectRegistry
                 _log.error("Exception while closing the RMI Registry: " + e.getMessage());
             }
         }
-        
+
         //ObjectName query to gather all Qpid related MBeans
         ObjectName mbeanNameQuery = null;
         try
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/plugins/PluginManager.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/plugins/PluginManager.java
index fe85b9d3e4..74abbccd2b 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/plugins/PluginManager.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/plugins/PluginManager.java
@@ -18,20 +18,20 @@
  */
 package org.apache.qpid.server.plugins;
 
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
 import org.apache.commons.configuration.ConfigurationException;
 import org.apache.felix.framework.Felix;
 import org.apache.felix.framework.util.StringMap;
 import org.apache.log4j.Logger;
-import org.apache.qpid.server.security.auth.manager.AnonymousAuthenticationManager;
-import org.apache.qpid.server.security.auth.manager.KerberosAuthenticationManager;
-import org.apache.qpid.server.security.auth.manager.SimpleLDAPAuthenticationManager;
-import org.osgi.framework.BundleActivator;
-import org.osgi.framework.BundleContext;
-import org.osgi.framework.BundleException;
-import org.osgi.framework.Version;
-import org.osgi.framework.launch.Framework;
-import org.osgi.util.tracker.ServiceTracker;
-
 import org.apache.qpid.common.Closeable;
 import org.apache.qpid.common.QpidProperties;
 import org.apache.qpid.server.configuration.TopicConfiguration;
@@ -43,24 +43,23 @@ import org.apache.qpid.server.exchange.ExchangeType;
 import org.apache.qpid.server.security.SecurityManager;
 import org.apache.qpid.server.security.SecurityPluginFactory;
 import org.apache.qpid.server.security.access.plugins.LegacyAccess;
+import org.apache.qpid.server.security.auth.manager.AnonymousAuthenticationManager;
 import org.apache.qpid.server.security.auth.manager.AuthenticationManagerPluginFactory;
+import org.apache.qpid.server.security.auth.manager.ExternalAuthenticationManager;
+import org.apache.qpid.server.security.auth.manager.KerberosAuthenticationManager;
 import org.apache.qpid.server.security.auth.manager.PrincipalDatabaseAuthenticationManager;
+import org.apache.qpid.server.security.auth.manager.SimpleLDAPAuthenticationManager;
 import org.apache.qpid.server.virtualhost.plugins.SlowConsumerDetection;
 import org.apache.qpid.server.virtualhost.plugins.VirtualHostPluginFactory;
 import org.apache.qpid.server.virtualhost.plugins.policies.TopicDeletePolicy;
 import org.apache.qpid.slowconsumerdetection.policies.SlowConsumerPolicyPluginFactory;
 import org.apache.qpid.util.FileUtils;
-
-import java.io.File;
-import java.io.IOException;
-import java.io.InputStream;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.IdentityHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
+import org.osgi.framework.BundleActivator;
+import org.osgi.framework.BundleContext;
+import org.osgi.framework.BundleException;
+import org.osgi.framework.Version;
+import org.osgi.framework.launch.Framework;
+import org.osgi.util.tracker.ServiceTracker;
 
 import static org.apache.felix.framework.util.FelixConstants.SYSTEMBUNDLE_ACTIVATORS_PROP;
 import static org.apache.felix.main.AutoProcessor.AUTO_DEPLOY_ACTION_PROPERY;
@@ -103,18 +102,18 @@ public class PluginManager implements Closeable
 
     /** The default name of the OSGI system package list. */
     private static final String DEFAULT_RESOURCE_NAME = "org/apache/qpid/server/plugins/OsgiSystemPackages.properties";
-    
+
     /** The name of the override system property that holds the name of the OSGI system package list. */
     private static final String FILE_PROPERTY = "qpid.osgisystempackages.properties";
-    
+
     private static final String OSGI_SYSTEM_PACKAGES;
-    
-    static 
+
+    static
     {
         final String filename = System.getProperty(FILE_PROPERTY);
         final InputStream is = FileUtils.openFileOrDefaultResource(filename, DEFAULT_RESOURCE_NAME,
                     PluginManager.class.getClassLoader());
-        
+
         try
         {
             Version qpidReleaseVersion;
@@ -126,14 +125,14 @@ public class PluginManager implements Closeable
             {
                 qpidReleaseVersion = null;
             }
-            
+
             final Properties p  = new Properties();
             p.load(is);
-            
+
             final OsgiSystemPackageUtil osgiSystemPackageUtil = new OsgiSystemPackageUtil(qpidReleaseVersion, (Map)p);
-            
+
             OSGI_SYSTEM_PACKAGES = osgiSystemPackageUtil.getFormattedSystemPackageString();
-            
+
             _logger.debug("List of OSGi system packages to be added: " + OSGI_SYSTEM_PACKAGES);
         }
         catch (IOException e)
@@ -142,8 +141,8 @@ public class PluginManager implements Closeable
             throw new ExceptionInInitializerError(e);
         }
     }
-    
-    
+
+
     public PluginManager(String pluginPath, String cachePath, BundleContext bundleContext) throws Exception
     {
         // Store all non-OSGi plugins
@@ -162,7 +161,9 @@ public class PluginManager implements Closeable
                 PrincipalDatabaseAuthenticationManager.PrincipalDatabaseAuthenticationManagerConfiguration.FACTORY,
                 AnonymousAuthenticationManager.AnonymousAuthenticationManagerConfiguration.FACTORY,
                 KerberosAuthenticationManager.KerberosAuthenticationManagerConfiguration.FACTORY,
-                SimpleLDAPAuthenticationManager.SimpleLDAPAuthenticationManagerConfiguration.FACTORY))
+                SimpleLDAPAuthenticationManager.SimpleLDAPAuthenticationManagerConfiguration.FACTORY,
+                ExternalAuthenticationManager.ExternalAuthenticationManagerConfiguration.FACTORY
+                ))
         {
             _configPlugins.put(configFactory.getParentPaths(), configFactory);
         }
@@ -179,7 +180,8 @@ public class PluginManager implements Closeable
 
         for (AuthenticationManagerPluginFactory<? extends Plugin> pluginFactory : Arrays.asList(
                 PrincipalDatabaseAuthenticationManager.FACTORY, AnonymousAuthenticationManager.FACTORY,
-                KerberosAuthenticationManager.FACTORY, SimpleLDAPAuthenticationManager.FACTORY))
+                KerberosAuthenticationManager.FACTORY, SimpleLDAPAuthenticationManager.FACTORY,
+                ExternalAuthenticationManager.FACTORY))
         {
             _authenticationManagerPlugins.put(pluginFactory.getPluginName(), pluginFactory);
         }
@@ -272,7 +274,7 @@ public class PluginManager implements Closeable
         _virtualHostTracker = new ServiceTracker(bundleContext, VirtualHostPluginFactory.class.getName(), null);
         _virtualHostTracker.open();
         _trackers.add(_virtualHostTracker);
- 
+
         _policyTracker = new ServiceTracker(bundleContext, SlowConsumerPolicyPluginFactory.class.getName(), null);
         _policyTracker.open();
         _trackers.add(_policyTracker);
@@ -285,9 +287,9 @@ public class PluginManager implements Closeable
     }
 
     private static <T> Map<String, T> getServices(ServiceTracker tracker)
-    {   
+    {
         Map<String, T> services = new HashMap<String, T>();
-        
+
         if ((tracker != null) && (tracker.getServices() != null))
         {
             for (Object service : tracker.getServices())
@@ -307,16 +309,16 @@ public class PluginManager implements Closeable
     }
 
     public static <T> Map<String, T> getServices(ServiceTracker tracker, Map<String, T> plugins)
-    {   
+    {
         Map<String, T> services = getServices(tracker);
         services.putAll(plugins);
         return services;
     }
 
     public Map<List<String>, ConfigurationPluginFactory> getConfigurationPlugins()
-    {   
+    {
         Map<List<String>, ConfigurationPluginFactory> services = new IdentityHashMap<List<String>, ConfigurationPluginFactory>();
-        
+
         if (_configTracker != null && _configTracker.getServices() != null)
         {
             for (Object service : _configTracker.getServices())
@@ -325,19 +327,19 @@ public class PluginManager implements Closeable
                 services.put(factory.getParentPaths(), factory);
             }
         }
-        
+
         services.putAll(_configPlugins);
 
         return services;
     }
 
     public Map<String, VirtualHostPluginFactory> getVirtualHostPlugins()
-    {   
+    {
         return getServices(_virtualHostTracker, _vhostPlugins);
     }
 
     public Map<String, SlowConsumerPolicyPluginFactory> getSlowConsumerPlugins()
-    {   
+    {
         return getServices(_policyTracker, _policyPlugins);
     }
 
@@ -345,7 +347,7 @@ public class PluginManager implements Closeable
     {
         return getServices(_exchangeTracker);
     }
-    
+
     public Map<String, SecurityPluginFactory> getSecurityPlugins()
     {
         return getServices(_securityTracker, _securityPlugins);
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
index 849aa05099..5db336649f 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolEngine.java
@@ -43,23 +43,7 @@ import org.apache.qpid.AMQConnectionException;
 import org.apache.qpid.AMQException;
 import org.apache.qpid.AMQSecurityException;
 import org.apache.qpid.codec.AMQCodecFactory;
-import org.apache.qpid.framing.AMQBody;
-import org.apache.qpid.framing.AMQDataBlock;
-import org.apache.qpid.framing.AMQFrame;
-import org.apache.qpid.framing.AMQMethodBody;
-import org.apache.qpid.framing.AMQProtocolHeaderException;
-import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.ChannelCloseBody;
-import org.apache.qpid.framing.ChannelCloseOkBody;
-import org.apache.qpid.framing.ConnectionCloseBody;
-import org.apache.qpid.framing.ContentBody;
-import org.apache.qpid.framing.ContentHeaderBody;
-import org.apache.qpid.framing.FieldTable;
-import org.apache.qpid.framing.HeartbeatBody;
-import org.apache.qpid.framing.MethodDispatcher;
-import org.apache.qpid.framing.MethodRegistry;
-import org.apache.qpid.framing.ProtocolInitiation;
-import org.apache.qpid.framing.ProtocolVersion;
+import org.apache.qpid.framing.*;
 import org.apache.qpid.properties.ConnectionStartProperties;
 import org.apache.qpid.protocol.AMQConstant;
 import org.apache.qpid.protocol.AMQMethodEvent;
@@ -84,7 +68,6 @@ import org.apache.qpid.server.output.ProtocolOutputConverter;
 import org.apache.qpid.server.output.ProtocolOutputConverterRegistry;
 import org.apache.qpid.server.queue.QueueEntry;
 import org.apache.qpid.server.registry.ApplicationRegistry;
-import org.apache.qpid.server.security.auth.sasl.UsernamePrincipal;
 import org.apache.qpid.server.state.AMQState;
 import org.apache.qpid.server.state.AMQStateManager;
 import org.apache.qpid.server.stats.StatisticsCounter;
@@ -1064,7 +1047,7 @@ public class AMQProtocolEngine implements ServerProtocolEngine, Managable, AMQPr
 
     public Principal getAuthorizedPrincipal()
     {
-        return _authorizedSubject == null ? null : UsernamePrincipal.getUsernamePrincipalFromSubject(_authorizedSubject);
+        return _authorizedSubject == null ? null : _authorizedSubject.getPrincipals().iterator().next();
     }
 
     public SocketAddress getRemoteAddress()
@@ -1077,6 +1060,11 @@ public class AMQProtocolEngine implements ServerProtocolEngine, Managable, AMQPr
         return _network.getLocalAddress();
     }
 
+    public Principal getPeerPrincipal()
+    {
+        return _network.getPeerPrincipal();
+    }
+
     public MethodRegistry getMethodRegistry()
     {
         return _methodRegistry;
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java
index f9bee93dbf..e833069320 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSession.java
@@ -21,6 +21,7 @@
 package org.apache.qpid.server.protocol;
 
 import java.net.SocketAddress;
+import java.security.Principal;
 import java.util.List;
 
 import javax.security.auth.Subject;
@@ -218,4 +219,6 @@ public interface AMQProtocolSession extends AMQVersionAwareProtocolSession, Auth
 
     void mgmtCloseChannel(int channelId);
 
+    public Principal getPeerPrincipal();
+
 }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java
index 182ef1ed82..0312db5dde 100755
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_0_10.java
@@ -88,7 +88,7 @@ public class ProtocolEngine_0_10  extends InputHandler implements ServerProtocol
         _network = network;
 
         _connection.setSender(new Disassembler(sender, MAX_FRAME_SIZE));
-
+        _connection.setPeerPrincipal(_network.getPeerPrincipal());
         // FIXME Two log messages to maintain compatibility with earlier protocol versions
         _connection.getLogActor().message(ConnectionMessages.OPEN(null, null, null, false, false, false));
         _connection.getLogActor().message(ConnectionMessages.OPEN(null, "0-10", null, false, true, false));
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0.java
index 045eafeba2..0b8bdff5c9 100755
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0.java
@@ -26,12 +26,13 @@ import java.util.UUID;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.logging.Level;
 import java.util.logging.Logger;
-import javax.security.auth.callback.CallbackHandler;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
 import org.apache.qpid.amqp_1_0.codec.FrameWriter;
 import org.apache.qpid.amqp_1_0.framing.AMQFrame;
 import org.apache.qpid.amqp_1_0.framing.FrameHandler;
 import org.apache.qpid.amqp_1_0.framing.OversizeFrameException;
-import org.apache.qpid.amqp_1_0.transport.CallbackHandlerSource;
+import org.apache.qpid.amqp_1_0.transport.SaslServerProvider;
 import org.apache.qpid.amqp_1_0.transport.ConnectionEndpoint;
 import org.apache.qpid.amqp_1_0.transport.Container;
 import org.apache.qpid.amqp_1_0.transport.FrameOutputHandler;
@@ -95,7 +96,7 @@ public class ProtocolEngine_1_0_0 implements ServerProtocolEngine, FrameOutputHa
     }
 
     private State _state = State.A;
-    
+
 
 
     public ProtocolEngine_1_0_0(final IApplicationRegistry appRegistry, long id)
@@ -143,7 +144,8 @@ public class ProtocolEngine_1_0_0 implements ServerProtocolEngine, FrameOutputHa
 
         Container container = new Container(_appRegistry.getBrokerId().toString());
 
-        _conn = new ConnectionEndpoint(container,asCallbackHandlerSource(_appRegistry.getAuthenticationManager(getLocalAddress())));
+        _conn = new ConnectionEndpoint(container, asSaslServerProvider(_appRegistry.getAuthenticationManager(
+                getLocalAddress())));
         _conn.setConnectionEventListener(new Connection_1_0(_appRegistry));
         _conn.setFrameOutputHandler(this);
         _conn.setRemoteAddress(_network.getRemoteAddress());
@@ -155,14 +157,14 @@ public class ProtocolEngine_1_0_0 implements ServerProtocolEngine, FrameOutputHa
         _sender.flush();
     }
 
-    private CallbackHandlerSource asCallbackHandlerSource(final AuthenticationManager authenticationManager)
+    private SaslServerProvider asSaslServerProvider(final AuthenticationManager authenticationManager)
     {
-        return new CallbackHandlerSource()
+        return new SaslServerProvider()
         {
             @Override
-            public CallbackHandler getHandler(String mechanism)
+            public SaslServer getSaslServer(String mechanism, String fqdn) throws SaslException
             {
-                return authenticationManager.getHandler(mechanism);
+                return authenticationManager.createSaslServer(mechanism, fqdn, null);
             }
         };
     }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0_SASL.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0_SASL.java
index 5d03567e03..876a8eb275 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0_SASL.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0_SASL.java
@@ -26,13 +26,14 @@ import java.nio.ByteBuffer;
 import java.util.UUID;
 import java.util.logging.Level;
 import java.util.logging.Logger;
-import javax.security.auth.callback.CallbackHandler;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
 import org.apache.qpid.amqp_1_0.codec.FrameWriter;
 import org.apache.qpid.amqp_1_0.codec.ProtocolHandler;
 import org.apache.qpid.amqp_1_0.framing.AMQFrame;
 import org.apache.qpid.amqp_1_0.framing.OversizeFrameException;
 import org.apache.qpid.amqp_1_0.framing.SASLFrameHandler;
-import org.apache.qpid.amqp_1_0.transport.CallbackHandlerSource;
+import org.apache.qpid.amqp_1_0.transport.SaslServerProvider;
 import org.apache.qpid.amqp_1_0.transport.ConnectionEndpoint;
 import org.apache.qpid.amqp_1_0.transport.Container;
 import org.apache.qpid.amqp_1_0.transport.FrameOutputHandler;
@@ -57,7 +58,7 @@ public class ProtocolEngine_1_0_0_SASL implements ServerProtocolEngine, FrameOut
        private long _createTime = System.currentTimeMillis();
        private ConnectionEndpoint _conn;
        private long _connectionId;
-    
+
        private static final ByteBuffer HEADER =
                ByteBuffer.wrap(new byte[]
                        {
@@ -163,8 +164,8 @@ public class ProtocolEngine_1_0_0_SASL implements ServerProtocolEngine, FrameOut
 
         Container container = new Container(_appRegistry.getBrokerId().toString());
 
-        _conn = new ConnectionEndpoint(container, asCallbackHandlerSource(ApplicationRegistry.getInstance()
-                                                                                             .getAuthenticationManager(getLocalAddress())));
+        _conn = new ConnectionEndpoint(container, asSaslServerProvider(ApplicationRegistry.getInstance()
+                .getAuthenticationManager(getLocalAddress())));
         _conn.setConnectionEventListener(new Connection_1_0(_appRegistry));
         _conn.setRemoteAddress(getRemoteAddress());
 
@@ -200,14 +201,14 @@ public class ProtocolEngine_1_0_0_SASL implements ServerProtocolEngine, FrameOut
 
     }
 
-    private CallbackHandlerSource asCallbackHandlerSource(final AuthenticationManager authenticationManager)
+    private SaslServerProvider asSaslServerProvider(final AuthenticationManager authenticationManager)
     {
-        return new CallbackHandlerSource()
+        return new SaslServerProvider()
         {
             @Override
-            public CallbackHandler getHandler(String mechanism)
+            public SaslServer getSaslServer(String mechanism, String fqdn) throws SaslException
             {
-                return authenticationManager.getHandler(mechanism);
+                return authenticationManager.createSaslServer(mechanism, fqdn, null);
             }
         };
     }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/AnonymousAuthenticationManager.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/AnonymousAuthenticationManager.java
index 1a6515f71f..0eb3963865 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/AnonymousAuthenticationManager.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/AnonymousAuthenticationManager.java
@@ -139,7 +139,7 @@ public class AnonymousAuthenticationManager implements AuthenticationManager
     }
 
     @Override
-    public SaslServer createSaslServer(String mechanism, String localFQDN) throws SaslException
+    public SaslServer createSaslServer(String mechanism, String localFQDN, Principal externalPrincipal) throws SaslException
     {
         if(ANONYMOUS.equals(mechanism))
         {
@@ -181,19 +181,6 @@ public class AnonymousAuthenticationManager implements AuthenticationManager
     }
 
     @Override
-    public CallbackHandler getHandler(String mechanism)
-    {
-        if(ANONYMOUS.equals(mechanism))
-        {
-            return _callbackHandler;
-        }
-        else
-        {
-            return null;
-        }
-    }
-
-    @Override
     public void close()
     {
     }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/AuthenticationManager.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/AuthenticationManager.java
index 6c1a917d5b..ccddcb7669 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/AuthenticationManager.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/AuthenticationManager.java
@@ -7,9 +7,9 @@
  * 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
@@ -20,25 +20,24 @@
  */
 package org.apache.qpid.server.security.auth.manager;
 
+import java.security.Principal;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
 import org.apache.qpid.common.Closeable;
 import org.apache.qpid.server.plugins.Plugin;
 import org.apache.qpid.server.security.auth.AuthenticationResult;
 
-import javax.security.auth.callback.CallbackHandler;
-import javax.security.sasl.SaslException;
-import javax.security.sasl.SaslServer;
-
 /**
  * Implementations of the AuthenticationManager are responsible for determining
  * the authenticity of a user's credentials.
- * 
+ *
  * If the authentication is successful, the manager is responsible for producing a populated
  * {@link javax.security.auth.Subject} containing the user's identity and zero or more principals representing
  * groups to which the user belongs.
  * <p>
  * The {@link #initialise()} method is responsible for registering SASL mechanisms required by
  * the manager.  The {@link #close()} method must reverse this registration.
- * 
+ *
  */
 public interface AuthenticationManager extends Closeable, Plugin
 {
@@ -64,11 +63,11 @@ public interface AuthenticationManager extends Closeable, Plugin
      *
      * @param mechanism mechanism name
      * @param localFQDN domain name
-     *
+     * @param externalPrincipal externally authenticated Principal
      * @return SASL server
      * @throws SaslException
      */
-    SaslServer createSaslServer(String mechanism, String localFQDN) throws SaslException;
+    SaslServer createSaslServer(String mechanism, String localFQDN, Principal externalPrincipal) throws SaslException;
 
     /**
      * Authenticates a user using SASL negotiation.
@@ -90,5 +89,4 @@ public interface AuthenticationManager extends Closeable, Plugin
      */
     AuthenticationResult authenticate(String username, String password);
 
-    CallbackHandler getHandler(String mechanism);
 }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/ExternalAuthenticationManager.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/ExternalAuthenticationManager.java
new file mode 100644
index 0000000000..2d6866b657
--- /dev/null
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/ExternalAuthenticationManager.java
@@ -0,0 +1,177 @@
+/*
+ * 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.server.security.auth.manager;
+
+import java.security.Principal;
+import java.util.Arrays;
+import java.util.List;
+import javax.security.auth.Subject;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.log4j.Logger;
+import org.apache.qpid.server.configuration.plugins.ConfigurationPlugin;
+import org.apache.qpid.server.configuration.plugins.ConfigurationPluginFactory;
+import org.apache.qpid.server.security.auth.AuthenticationResult;
+import org.apache.qpid.server.security.auth.sasl.external.ExternalSaslServer;
+
+public class ExternalAuthenticationManager implements AuthenticationManager
+{
+    private static final Logger _logger = Logger.getLogger(ExternalAuthenticationManager.class);
+
+    private static final String EXTERNAL = "EXTERNAL";
+
+    static final ExternalAuthenticationManager INSTANCE = new ExternalAuthenticationManager();
+
+    public static class ExternalAuthenticationManagerConfiguration extends ConfigurationPlugin
+    {
+
+        public static final ConfigurationPluginFactory FACTORY =
+                new ConfigurationPluginFactory()
+                {
+                    public List<String> getParentPaths()
+                    {
+                        return Arrays.asList("security.external-auth-manager");
+                    }
+
+                    public ConfigurationPlugin newInstance(final String path, final Configuration config) throws ConfigurationException
+                    {
+                        final ConfigurationPlugin instance = new ExternalAuthenticationManagerConfiguration();
+
+                        instance.setConfiguration(path, config);
+                        return instance;
+                    }
+                };
+
+        public String[] getElementsProcessed()
+        {
+            return new String[0];
+        }
+
+        public void validateConfiguration() throws ConfigurationException
+        {
+        }
+
+        }
+
+
+    public static final AuthenticationManagerPluginFactory<ExternalAuthenticationManager> FACTORY = new AuthenticationManagerPluginFactory<ExternalAuthenticationManager>()
+    {
+        public ExternalAuthenticationManager newInstance(final ConfigurationPlugin config) throws ConfigurationException
+        {
+            ExternalAuthenticationManagerConfiguration configuration =
+                    config == null
+                            ? null
+                            : (ExternalAuthenticationManagerConfiguration) config.getConfiguration(ExternalAuthenticationManagerConfiguration.class.getName());
+
+            // If there is no configuration for this plugin then don't load it.
+            if (configuration == null)
+            {
+                _logger.info("No authentication-manager configuration found for ExternalAuthenticationManager");
+                return null;
+            }
+            return INSTANCE;
+        }
+
+        public Class<ExternalAuthenticationManager> getPluginClass()
+        {
+            return ExternalAuthenticationManager.class;
+        }
+
+        public String getPluginName()
+        {
+            return ExternalAuthenticationManager.class.getName();
+        }
+    };
+
+
+    private ExternalAuthenticationManager()
+    {
+    }
+
+    @Override
+    public void initialise()
+    {
+
+    }
+
+    @Override
+    public String getMechanisms()
+    {
+        return EXTERNAL;
+    }
+
+    @Override
+    public SaslServer createSaslServer(String mechanism, String localFQDN, Principal externalPrincipal) throws SaslException
+    {
+        if(EXTERNAL.equals(mechanism))
+        {
+            return new ExternalSaslServer(externalPrincipal);
+        }
+        else
+        {
+            throw new SaslException("Unknown mechanism: " + mechanism);
+        }
+    }
+
+    @Override
+    public AuthenticationResult authenticate(SaslServer server, byte[] response)
+    {
+        // Process response from the client
+        try
+        {
+            byte[] challenge = server.evaluateResponse(response != null ? response : new byte[0]);
+
+            Principal principal = ((ExternalSaslServer)server).getAuthenticatedPrincipal();
+
+            if(principal != null)
+            {
+                final Subject subject = new Subject();
+                subject.getPrincipals().add(principal);
+                return new AuthenticationResult(subject);
+            }
+            else
+            {
+                return new AuthenticationResult(AuthenticationResult.AuthenticationStatus.ERROR);
+            }
+        }
+        catch (SaslException e)
+        {
+            return new AuthenticationResult(AuthenticationResult.AuthenticationStatus.ERROR,e);
+        }
+
+    }
+
+    @Override
+    public AuthenticationResult authenticate(String username, String password)
+    {
+        return new AuthenticationResult(AuthenticationResult.AuthenticationStatus.ERROR);
+    }
+
+    @Override
+    public void close()
+    {
+    }
+
+    @Override
+    public void configure(ConfigurationPlugin config) throws ConfigurationException
+    {
+    }
+}
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/KerberosAuthenticationManager.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/KerberosAuthenticationManager.java
index 21c9a85bf7..d735ecb1d4 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/KerberosAuthenticationManager.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/KerberosAuthenticationManager.java
@@ -19,6 +19,7 @@
 package org.apache.qpid.server.security.auth.manager;
 
 import java.io.IOException;
+import java.security.Principal;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
@@ -126,7 +127,7 @@ public class KerberosAuthenticationManager implements AuthenticationManager
     }
 
     @Override
-    public SaslServer createSaslServer(String mechanism, String localFQDN) throws SaslException
+    public SaslServer createSaslServer(String mechanism, String localFQDN, Principal externalPrincipal) throws SaslException
     {
         if(GSSAPI_MECHANISM.equals(mechanism))
         {
@@ -181,19 +182,6 @@ public class KerberosAuthenticationManager implements AuthenticationManager
     }
 
     @Override
-    public CallbackHandler getHandler(String mechanism)
-    {
-        if(GSSAPI_MECHANISM.equals(mechanism))
-        {
-            return _callbackHandler;
-        }
-        else
-        {
-            return null;
-        }
-    }
-
-    @Override
     public void close()
     {
     }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/PrincipalDatabaseAuthenticationManager.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/PrincipalDatabaseAuthenticationManager.java
index b34e6acc6d..24b365d34c 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/PrincipalDatabaseAuthenticationManager.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/PrincipalDatabaseAuthenticationManager.java
@@ -14,12 +14,13 @@
  *  "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.    
+ *  under the License.
+ *
  *
- * 
  */
 package org.apache.qpid.server.security.auth.manager;
 
+import java.security.Principal;
 import org.apache.commons.configuration.Configuration;
 import org.apache.commons.configuration.ConfigurationException;
 import org.apache.log4j.Logger;
@@ -60,9 +61,9 @@ import java.util.TreeMap;
  * Concrete implementation of the AuthenticationManager that determines if supplied
  * user credentials match those appearing in a PrincipalDatabase.   The implementation
  * of the PrincipalDatabase is determined from the configuration.
- * 
+ *
  * This implementation also registers the JMX UserManagemement MBean.
- * 
+ *
  * This plugin expects configuration such as:
  *
  * <pre>
@@ -133,7 +134,7 @@ public class PrincipalDatabaseAuthenticationManager implements AuthenticationMan
     };
 
     public static class PrincipalDatabaseAuthenticationManagerConfiguration extends ConfigurationPlugin {
- 
+
         public static final ConfigurationPluginFactory FACTORY = new ConfigurationPluginFactory()
         {
             public List<String> getParentPaths()
@@ -144,7 +145,7 @@ public class PrincipalDatabaseAuthenticationManager implements AuthenticationMan
             public ConfigurationPlugin newInstance(final String path, final Configuration config) throws ConfigurationException
             {
                 final ConfigurationPlugin instance = new PrincipalDatabaseAuthenticationManagerConfiguration();
-                
+
                 instance.setConfiguration(path, config);
                 return instance;
             }
@@ -160,12 +161,12 @@ public class PrincipalDatabaseAuthenticationManager implements AuthenticationMan
         public void validateConfiguration() throws ConfigurationException
         {
         }
-  
+
         public String getPrincipalDatabaseClass()
         {
             return getConfig().getString("principal-database.class");
         }
-  
+
         public Map<String,String> getPdClassAttributeMap() throws ConfigurationException
         {
             final List<String> argumentNames = (List) getConfig().getList("principal-database.attributes.attribute.name");
@@ -184,7 +185,7 @@ public class PrincipalDatabaseAuthenticationManager implements AuthenticationMan
         }
     }
 
-    protected PrincipalDatabaseAuthenticationManager()  
+    protected PrincipalDatabaseAuthenticationManager()
     {
     }
 
@@ -214,7 +215,7 @@ public class PrincipalDatabaseAuthenticationManager implements AuthenticationMan
         registerManagement();
     }
 
-    private void initialiseAuthenticationMechanisms(Map<String, Class<? extends SaslServerFactory>> providerMap, PrincipalDatabase database) 
+    private void initialiseAuthenticationMechanisms(Map<String, Class<? extends SaslServerFactory>> providerMap, PrincipalDatabase database)
     {
         if (database == null || database.getMechanisms().size() == 0)
         {
@@ -262,7 +263,7 @@ public class PrincipalDatabaseAuthenticationManager implements AuthenticationMan
 
         _principalDatabase = createPrincipalDatabaseImpl(pdClazz);
 
-        configPrincipalDatabase(_principalDatabase, pdamConfig);        
+        configPrincipalDatabase(_principalDatabase, pdamConfig);
     }
 
     public String getMechanisms()
@@ -270,7 +271,7 @@ public class PrincipalDatabaseAuthenticationManager implements AuthenticationMan
         return _mechanisms;
     }
 
-    public SaslServer createSaslServer(String mechanism, String localFQDN) throws SaslException
+    public SaslServer createSaslServer(String mechanism, String localFQDN, Principal externalPrincipal) throws SaslException
     {
         return Sasl.createSaslServer(mechanism, "AMQP", localFQDN, _serverCreationProperties.get(mechanism),
                                      _callbackHandlerMap.get(mechanism));
@@ -303,11 +304,6 @@ public class PrincipalDatabaseAuthenticationManager implements AuthenticationMan
         }
     }
 
-    public CallbackHandler getHandler(String mechanism)
-    {
-        return _callbackHandlerMap.get(mechanism);
-    }
-
     /**
      * @see org.apache.qpid.server.security.auth.manager.AuthenticationManager#authenticate(String, String)
      */
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/SimpleLDAPAuthenticationManager.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/SimpleLDAPAuthenticationManager.java
index c41e814739..64b24e28bc 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/SimpleLDAPAuthenticationManager.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/manager/SimpleLDAPAuthenticationManager.java
@@ -20,6 +20,7 @@
 package org.apache.qpid.server.security.auth.manager;
 
 import java.io.IOException;
+import java.security.Principal;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Hashtable;
@@ -180,7 +181,7 @@ public class SimpleLDAPAuthenticationManager implements AuthenticationManager
     }
 
     @Override
-    public SaslServer createSaslServer(String mechanism, String localFQDN) throws SaslException
+    public SaslServer createSaslServer(String mechanism, String localFQDN, Principal externalPrincipal) throws SaslException
     {
         if(PLAIN_MECHANISM.equals(mechanism))
         {
@@ -254,19 +255,6 @@ public class SimpleLDAPAuthenticationManager implements AuthenticationManager
     }
 
     @Override
-    public CallbackHandler getHandler(String mechanism)
-    {
-        if(PLAIN_MECHANISM.equals(mechanism))
-        {
-            return new PlainCallbackHandler();
-        }
-        else
-        {
-            return null;
-        }
-    }
-
-    @Override
     public void close()
     {
     }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/external/ExternalSaslServer.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/external/ExternalSaslServer.java
new file mode 100644
index 0000000000..9c2bca2d0b
--- /dev/null
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/security/auth/sasl/external/ExternalSaslServer.java
@@ -0,0 +1,82 @@
+/*
+ * 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.server.security.auth.sasl.external;
+
+import java.security.Principal;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
+
+
+public class ExternalSaslServer implements SaslServer
+{
+    public static final String MECHANISM = "EXTERNAL";
+
+    private boolean _complete = false;
+    private final Principal _externalPrincipal;
+
+    public ExternalSaslServer(Principal externalPrincipal)
+    {
+        _externalPrincipal = externalPrincipal;
+    }
+
+    public String getMechanismName()
+    {
+        return MECHANISM;
+    }
+
+    public byte[] evaluateResponse(byte[] response) throws SaslException
+    {
+        _complete = true;
+        return null;
+    }
+
+    public boolean isComplete()
+    {
+        return _complete;
+    }
+
+    public String getAuthorizationID()
+    {
+        return null;
+    }
+
+    public byte[] unwrap(byte[] incoming, int offset, int len) throws SaslException
+    {
+        throw new SaslException("Unsupported operation");
+    }
+
+    public byte[] wrap(byte[] outgoing, int offset, int len) throws SaslException
+    {
+        throw new SaslException("Unsupported operation");
+    }
+
+    public Object getNegotiatedProperty(String propName)
+    {
+        return null;
+    }
+
+    public void dispose() throws SaslException
+    {
+    }
+
+    public Principal getAuthenticatedPrincipal()
+    {
+        return _externalPrincipal;
+    }
+}
\ No newline at end of file
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java
index 77d07e49f3..c9482b9712 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnection.java
@@ -20,6 +20,15 @@
  */
 package org.apache.qpid.server.transport;
 
+import java.security.Principal;
+import java.text.MessageFormat;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import javax.management.JMException;
+import javax.security.auth.Subject;
 import org.apache.qpid.AMQException;
 import org.apache.qpid.protocol.AMQConstant;
 import org.apache.qpid.server.configuration.ConnectionConfig;
@@ -33,7 +42,6 @@ import org.apache.qpid.server.management.ManagedObject;
 import org.apache.qpid.server.protocol.AMQConnectionModel;
 import org.apache.qpid.server.protocol.AMQSessionModel;
 import org.apache.qpid.server.security.AuthorizationHolder;
-import org.apache.qpid.server.security.auth.sasl.UsernamePrincipal;
 import org.apache.qpid.server.stats.StatisticsCounter;
 import org.apache.qpid.server.virtualhost.VirtualHost;
 import org.apache.qpid.transport.Connection;
@@ -48,16 +56,6 @@ import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.CONNECTIO
 import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.SOCKET_FORMAT;
 import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.USER_FORMAT;
 
-import javax.management.JMException;
-import javax.security.auth.Subject;
-import java.security.Principal;
-import java.text.MessageFormat;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.UUID;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicLong;
-
 public class ServerConnection extends Connection implements Managable, AMQConnectionModel, LogSubject, AuthorizationHolder
 {
     private ConnectionConfig _config;
@@ -75,6 +73,7 @@ public class ServerConnection extends Connection implements Managable, AMQConnec
     private VirtualHost _virtualHost;
     private AtomicLong _lastIoTime = new AtomicLong();
     private boolean _blocking;
+    private Principal _peerPrincipal;
 
     public ServerConnection(final long connectionId)
     {
@@ -430,7 +429,7 @@ public class ServerConnection extends Connection implements Managable, AMQConnec
         else
         {
             _authorizedSubject = authorizedSubject;
-            _authorizedPrincipal = UsernamePrincipal.getUsernamePrincipalFromSubject(_authorizedSubject);
+            _authorizedPrincipal = authorizedSubject.getPrincipals().iterator().next();
         }
     }
 
@@ -539,4 +538,14 @@ public class ServerConnection extends Connection implements Managable, AMQConnec
     {
         return getConnectionDelegate().getClientVersion();
     }
+
+    public Principal getPeerPrincipal()
+    {
+        return _peerPrincipal;
+    }
+
+    public void setPeerPrincipal(Principal peerPrincipal)
+    {
+        _peerPrincipal = peerPrincipal;
+    }
 }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java
index a55d50cc54..ad59c56878 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java
@@ -20,14 +20,21 @@
  */
 package org.apache.qpid.server.transport;
 
-import static org.apache.qpid.transport.Connection.State.CLOSE_RCVD;
-
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.StringTokenizer;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
 import org.apache.qpid.common.ServerPropertyNames;
 import org.apache.qpid.properties.ConnectionStartProperties;
 import org.apache.qpid.protocol.ProtocolEngine;
 import org.apache.qpid.server.configuration.BrokerConfig;
 import org.apache.qpid.server.protocol.AMQConnectionModel;
-import org.apache.qpid.server.registry.ApplicationRegistry;
 import org.apache.qpid.server.registry.IApplicationRegistry;
 import org.apache.qpid.server.security.SecurityManager;
 import org.apache.qpid.server.security.auth.AuthenticationResult;
@@ -40,16 +47,7 @@ import org.apache.qpid.transport.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import javax.security.sasl.SaslException;
-import javax.security.sasl.SaslServer;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.StringTokenizer;
+import static org.apache.qpid.transport.Connection.State.CLOSE_RCVD;
 
 public class ServerConnectionDelegate extends ServerDelegate
 {
@@ -112,17 +110,16 @@ public class ServerConnectionDelegate extends ServerDelegate
         return ssn;
     }
 
-    protected SaslServer createSaslServer(String mechanism) throws SaslException
+    protected SaslServer createSaslServer(Connection conn, String mechanism) throws SaslException
     {
-        return _authManager.createSaslServer(mechanism, _localFQDN);
+        return _authManager.createSaslServer(mechanism, _localFQDN, ((ServerConnection) conn).getPeerPrincipal());
 
     }
 
     protected void secure(final SaslServer ss, final Connection conn, final byte[] response)
     {
-        final AuthenticationResult authResult = _authManager.authenticate(ss, response);
         final ServerConnection sconn = (ServerConnection) conn;
-
+        final AuthenticationResult authResult = _authManager.authenticate(ss, response);
 
         if (AuthenticationStatus.SUCCESS.equals(authResult.getStatus()))
         {
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/manager/AnonymousAuthenticationManagerTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/manager/AnonymousAuthenticationManagerTest.java
index eecde964a3..9dcd22c088 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/manager/AnonymousAuthenticationManagerTest.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/manager/AnonymousAuthenticationManagerTest.java
@@ -79,13 +79,13 @@ public class AnonymousAuthenticationManagerTest extends InternalBrokerBaseCase
 
     public void testCreateSaslServer() throws Exception
     {
-        SaslServer server = _manager.createSaslServer("ANONYMOUS", "example.example.com");
+        SaslServer server = _manager.createSaslServer("ANONYMOUS", "example.example.com", null);
 
         assertEquals("Sasl Server mechanism name is not as expected", "ANONYMOUS", server.getMechanismName());
 
         try
         {
-            server = _manager.createSaslServer("PLAIN", "example.example.com");
+            server = _manager.createSaslServer("PLAIN", "example.example.com", null);
             fail("Expected creating SaslServer with incorrect mechanism to throw an exception");
         }
         catch (SaslException e)
@@ -96,7 +96,7 @@ public class AnonymousAuthenticationManagerTest extends InternalBrokerBaseCase
 
     public void testAuthenticate() throws Exception
     {
-        SaslServer saslServer = _manager.createSaslServer("ANONYMOUS", "example.example.com");
+        SaslServer saslServer = _manager.createSaslServer("ANONYMOUS", "example.example.com", null);
         AuthenticationResult result = _manager.authenticate(saslServer, new byte[0]);
         assertNotNull(result);
         assertEquals("Expected authentication to be successful",
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/manager/ExternalAuthenticationManagerTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/manager/ExternalAuthenticationManagerTest.java
new file mode 100644
index 0000000000..c1a55ef2ad
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/manager/ExternalAuthenticationManagerTest.java
@@ -0,0 +1,120 @@
+/*
+ * 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.server.security.auth.manager;
+
+import javax.security.auth.x500.X500Principal;
+import javax.security.sasl.SaslException;
+import javax.security.sasl.SaslServer;
+import org.apache.commons.configuration.CompositeConfiguration;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.commons.configuration.XMLConfiguration;
+import org.apache.qpid.server.configuration.plugins.ConfigurationPlugin;
+import org.apache.qpid.server.security.auth.AuthenticationResult;
+import org.apache.qpid.server.security.auth.database.PlainPasswordFilePrincipalDatabase;
+import org.apache.qpid.server.util.InternalBrokerBaseCase;
+
+public class ExternalAuthenticationManagerTest extends InternalBrokerBaseCase
+{
+
+    private AuthenticationManager _manager = null;
+
+    public void setUp() throws Exception
+    {
+        _manager = ExternalAuthenticationManager.INSTANCE;
+    }
+
+
+    public void tearDown() throws Exception
+    {
+        if(_manager != null)
+        {
+            _manager = null;
+        }
+    }
+
+    private ConfigurationPlugin getPlainDatabaseConfig() throws ConfigurationException
+    {
+        final ConfigurationPlugin config = new PrincipalDatabaseAuthenticationManager.PrincipalDatabaseAuthenticationManagerConfiguration();
+
+        XMLConfiguration xmlconfig = new XMLConfiguration();
+        xmlconfig.addProperty("pd-auth-manager.principal-database.class", PlainPasswordFilePrincipalDatabase.class.getName());
+
+        // Create a CompositeConfiguration as this is what the broker uses
+        CompositeConfiguration composite = new CompositeConfiguration();
+        composite.addConfiguration(xmlconfig);
+        config.setConfiguration("security", xmlconfig);
+        return config;
+    }
+
+
+    public void testConfiguration() throws Exception
+    {
+        AuthenticationManager authenticationManager =
+                ExternalAuthenticationManager.FACTORY.newInstance(getPlainDatabaseConfig());
+
+        assertNull("ExternalAuthenticationManager unexpectedly created when not in config", authenticationManager);
+    }
+
+    public void testGetMechanisms() throws Exception
+    {
+        assertEquals("EXTERNAL", _manager.getMechanisms());
+    }
+
+    public void testCreateSaslServer() throws Exception
+    {
+        SaslServer server = _manager.createSaslServer("EXTERNAL", "example.example.com", null);
+
+        assertEquals("Sasl Server mechanism name is not as expected", "EXTERNAL", server.getMechanismName());
+
+        try
+        {
+            server = _manager.createSaslServer("PLAIN", "example.example.com", null);
+            fail("Expected creating SaslServer with incorrect mechanism to throw an exception");
+        }
+        catch (SaslException e)
+        {
+            // pass
+        }
+    }
+
+    public void testAuthenticate() throws Exception
+    {
+        X500Principal principal = new X500Principal("CN=person, DC=example, DC=com");
+        SaslServer saslServer = _manager.createSaslServer("EXTERNAL", "example.example.com", principal);
+
+        AuthenticationResult result = _manager.authenticate(saslServer, new byte[0]);
+        assertNotNull(result);
+        assertEquals("Expected authentication to be successful",
+                     AuthenticationResult.AuthenticationStatus.SUCCESS,
+                     result.getStatus());
+        assertEquals("Expected principal to be unchanged",
+                             principal,
+                             result.getSubject().getPrincipals().iterator().next());
+
+        saslServer = _manager.createSaslServer("EXTERNAL", "example.example.com", null);
+        result = _manager.authenticate(saslServer, new byte[0]);
+        assertNotNull(result);
+                assertEquals("Expected authentication to be unsuccessful",
+                             AuthenticationResult.AuthenticationStatus.ERROR,
+                             result.getStatus());
+
+    }
+
+
+}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/manager/PrincipalDatabaseAuthenticationManagerTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/manager/PrincipalDatabaseAuthenticationManagerTest.java
index 1a42fe3886..47c189e4fa 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/manager/PrincipalDatabaseAuthenticationManagerTest.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/manager/PrincipalDatabaseAuthenticationManagerTest.java
@@ -167,7 +167,7 @@ public class PrincipalDatabaseAuthenticationManagerTest extends InternalBrokerBa
      */
     public void testSaslMechanismCreation() throws Exception
     {
-        SaslServer server = _manager.createSaslServer("CRAM-MD5", "localhost");
+        SaslServer server = _manager.createSaslServer("CRAM-MD5", "localhost", null);
         assertNotNull(server);
         // Merely tests the creation of the mechanism. Mechanisms themselves are tested
         // by their own tests.
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/rmi/RMIPasswordAuthenticatorTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/rmi/RMIPasswordAuthenticatorTest.java
index df3bbb3e8b..f6675e917e 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/rmi/RMIPasswordAuthenticatorTest.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/security/auth/rmi/RMIPasswordAuthenticatorTest.java
@@ -20,6 +20,7 @@
  */
 package org.apache.qpid.server.security.auth.rmi;
 
+import java.security.Principal;
 import junit.framework.TestCase;
 
 import org.apache.qpid.server.configuration.plugins.ConfigurationPlugin;
@@ -29,7 +30,6 @@ import org.apache.qpid.server.security.auth.manager.AuthenticationManager;
 
 import javax.management.remote.JMXPrincipal;
 import javax.security.auth.Subject;
-import javax.security.auth.callback.CallbackHandler;
 import javax.security.sasl.SaslException;
 import javax.security.sasl.SaslServer;
 import java.util.Collections;
@@ -71,14 +71,14 @@ public class RMIPasswordAuthenticatorTest extends TestCase
                 newSubject.equals(expectedSubject));
 
     }
-    
+
     /**
      * Tests a unsuccessful authentication.
      */
     public void testUsernameOrPasswordInvalid()
     {
         _rmipa.setAuthenticationManager(createTestAuthenticationManager(false, null));
-        
+
         try
         {
             _rmipa.authenticate(_credentials);
@@ -166,7 +166,7 @@ public class RMIPasswordAuthenticatorTest extends TestCase
             assertEquals("Unexpected exception message",
                     RMIPasswordAuthenticator.SHOULD_HAVE_2_ELEMENTS, se.getMessage());
         }
-        
+
         // Test handling of null credentials
         try
         {
@@ -180,7 +180,7 @@ public class RMIPasswordAuthenticatorTest extends TestCase
             assertEquals("Unexpected exception message",
                     RMIPasswordAuthenticator.CREDENTIALS_REQUIRED, se.getMessage());
         }
-        
+
         try
         {
             //send a null password
@@ -193,7 +193,7 @@ public class RMIPasswordAuthenticatorTest extends TestCase
             assertEquals("Unexpected exception message",
                     RMIPasswordAuthenticator.SHOULD_BE_NON_NULL, se.getMessage());
         }
-        
+
         try
         {
             //send a null username
@@ -232,7 +232,7 @@ public class RMIPasswordAuthenticatorTest extends TestCase
                 throw new UnsupportedOperationException();
             }
 
-            public SaslServer createSaslServer(String mechanism, String localFQDN) throws SaslException
+            public SaslServer createSaslServer(String mechanism, String localFQDN, Principal externalPrincipal) throws SaslException
             {
                 throw new UnsupportedOperationException();
             }
@@ -257,10 +257,6 @@ public class RMIPasswordAuthenticatorTest extends TestCase
                 }
             }
 
-            public CallbackHandler getHandler(String mechanism)
-            {
-                return null;
-            }
         };
     }
 }
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/NetworkTransportConfiguration.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/NetworkTransportConfiguration.java
index 472beb6bb1..20d6f98fa6 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/transport/NetworkTransportConfiguration.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/NetworkTransportConfiguration.java
@@ -25,17 +25,17 @@ import java.net.InetSocketAddress;
 /**
  * This interface provides a means for NetworkDrivers to configure TCP options such as incoming and outgoing
  * buffer sizes and set particular options on the socket. NetworkDrivers should honour the values returned
- * from here if the underlying implementation supports them.  
- */ 
-public interface NetworkTransportConfiguration  
-{  
-    // Taken from Socket  
-    Boolean getTcpNoDelay(); 
+ * from here if the underlying implementation supports them.
+ */
+public interface NetworkTransportConfiguration
+{
+    // Taken from Socket
+    Boolean getTcpNoDelay();
 
-    // The amount of memory in bytes to allocate to the incoming buffer 
-    Integer getReceiveBufferSize();  
+    // The amount of memory in bytes to allocate to the incoming buffer
+    Integer getReceiveBufferSize();
 
-    // The amount of memory in bytes to allocate to the outgoing buffer 
+    // The amount of memory in bytes to allocate to the outgoing buffer
     Integer getSendBufferSize();
 
     Integer getPort();
@@ -47,4 +47,8 @@ public interface NetworkTransportConfiguration
     Integer getConnectorProcessors();
 
     InetSocketAddress getAddress();
+
+    boolean needClientAuth();
+
+    boolean wantClientAuth();
 }
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/ServerDelegate.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/ServerDelegate.java
index ec409d1c72..e9a7d51456 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/transport/ServerDelegate.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/ServerDelegate.java
@@ -78,7 +78,7 @@ public class ServerDelegate extends ConnectionDelegate
         try
         {
             
-            SaslServer ss = createSaslServer(mechanism);
+            SaslServer ss = createSaslServer(conn, mechanism);
             if (ss == null)
             {
                 conn.connectionClose(ConnectionCloseCode.CONNECTION_FORCED,
@@ -94,7 +94,7 @@ public class ServerDelegate extends ConnectionDelegate
         }
     }
 
-    protected SaslServer createSaslServer(String mechanism)
+    protected SaslServer createSaslServer(Connection conn, String mechanism)
             throws SaslException
     {
         SaslServer ss = Sasl.createSaslServer(mechanism, "AMQP", "localhost", null, null);
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/network/NetworkConnection.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/network/NetworkConnection.java
index 2cc7c14f00..12c42d6643 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/transport/network/NetworkConnection.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/network/NetworkConnection.java
@@ -20,10 +20,10 @@
  */
 package org.apache.qpid.transport.network;
 
-import org.apache.qpid.transport.Sender;
-
 import java.net.SocketAddress;
 import java.nio.ByteBuffer;
+import java.security.Principal;
+import org.apache.qpid.transport.Sender;
 
 public interface NetworkConnection
 {
@@ -46,4 +46,8 @@ public interface NetworkConnection
     void setMaxWriteIdle(int sec);
 
     void setMaxReadIdle(int sec);
-}
\ No newline at end of file
+
+    void setPeerPrincipal(Principal principal);
+
+    Principal getPeerPrincipal();
+}
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkConnection.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkConnection.java
index 4046691779..2658296c5f 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkConnection.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkConnection.java
@@ -20,16 +20,15 @@
  */
 package org.apache.qpid.transport.network.io;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.qpid.transport.Receiver;
-import org.apache.qpid.transport.Sender;
-import org.apache.qpid.transport.network.NetworkConnection;
-
 import java.net.Socket;
 import java.net.SocketAddress;
 import java.nio.ByteBuffer;
+import java.security.Principal;
+import org.apache.qpid.transport.Receiver;
+import org.apache.qpid.transport.Sender;
+import org.apache.qpid.transport.network.NetworkConnection;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class IoNetworkConnection implements NetworkConnection
 {
@@ -38,6 +37,7 @@ public class IoNetworkConnection implements NetworkConnection
     private final long _timeout;
     private final IoSender _ioSender;
     private final IoReceiver _ioReceiver;
+    private Principal _principal;
 
     public IoNetworkConnection(Socket socket, Receiver<ByteBuffer> delegate,
             int sendBufferSize, int receiveBufferSize, long timeout)
@@ -97,4 +97,16 @@ public class IoNetworkConnection implements NetworkConnection
         // TODO implement support for setting heartbeating config in this way
         // Currently a socket timeout is used in IoSender
     }
+
+    @Override
+    public void setPeerPrincipal(Principal principal)
+    {
+        _principal = principal;
+    }
+
+    @Override
+    public Principal getPeerPrincipal()
+    {
+        return _principal;
+    }
 }
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java
index 42c8334a5d..56f6989aae 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoNetworkTransport.java
@@ -27,10 +27,12 @@ import java.net.ServerSocket;
 import java.net.Socket;
 import java.net.SocketException;
 import java.nio.ByteBuffer;
-
+import java.security.Principal;
 import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLPeerUnverifiedException;
+import javax.net.ssl.SSLServerSocket;
 import javax.net.ssl.SSLServerSocketFactory;
-
+import javax.net.ssl.SSLSocket;
 import org.apache.qpid.protocol.ProtocolEngine;
 import org.apache.qpid.protocol.ProtocolEngineFactory;
 import org.apache.qpid.transport.ConnectionSettings;
@@ -167,6 +169,9 @@ public class IoNetworkTransport implements OutgoingNetworkTransport, IncomingNet
             {
                 SSLServerSocketFactory socketFactory = _sslContext.getServerSocketFactory();
                 _serverSocket = socketFactory.createServerSocket();
+                ((SSLServerSocket)_serverSocket).setNeedClientAuth(config.needClientAuth());
+                ((SSLServerSocket)_serverSocket).setWantClientAuth(config.wantClientAuth());
+
             }
 
             _serverSocket.setReuseAddress(true);
@@ -216,10 +221,24 @@ public class IoNetworkTransport implements OutgoingNetworkTransport, IncomingNet
                         socket.setSendBufferSize(sendBufferSize);
                         socket.setReceiveBufferSize(receiveBufferSize);
 
+
                         ProtocolEngine engine = _factory.newProtocolEngine();
 
                         NetworkConnection connection = new IoNetworkConnection(socket, engine, sendBufferSize, receiveBufferSize, _timeout);
 
+                        if(_sslContext != null)
+                        {
+                            try
+                            {
+                                Principal peerPrincipal = ((SSLSocket) socket).getSession().getPeerPrincipal();
+                                connection.setPeerPrincipal(peerPrincipal);
+                            }
+                            catch(SSLPeerUnverifiedException e)
+                            {
+                                // ignore
+                            }
+                        }
+
                         engine.setNetworkConnection(connection, connection.getSender());
 
                         connection.start();
diff --git a/qpid/java/common/src/test/java/org/apache/qpid/transport/TestNetworkConnection.java b/qpid/java/common/src/test/java/org/apache/qpid/transport/TestNetworkConnection.java
index 548e8dab12..893f66c5ff 100644
--- a/qpid/java/common/src/test/java/org/apache/qpid/transport/TestNetworkConnection.java
+++ b/qpid/java/common/src/test/java/org/apache/qpid/transport/TestNetworkConnection.java
@@ -20,6 +20,7 @@
  */
 package org.apache.qpid.transport;
 
+import java.security.Principal;
 import org.apache.qpid.protocol.ProtocolEngineFactory;
 import org.apache.qpid.ssl.SSLContextFactory;
 import org.apache.qpid.transport.network.NetworkConnection;
@@ -71,6 +72,17 @@ public class TestNetworkConnection implements NetworkConnection
 
     }
 
+    @Override
+    public void setPeerPrincipal(Principal principal)
+    {
+    }
+
+    @Override
+    public Principal getPeerPrincipal()
+    {
+        return null;
+    }
+
     public void setMaxWriteIdle(int idleTime)
     {
 
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/client/ssl/SSLTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/client/ssl/SSLTest.java
index 1cd088b736..39689f5096 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/client/ssl/SSLTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/client/ssl/SSLTest.java
@@ -46,6 +46,7 @@ public class SSLTest extends QpidBrokerTestCase
             setTestClientSystemProperty("profile.use_ssl", "true");
             setConfigurationProperty("connector.ssl.enabled", "true");
             setConfigurationProperty("connector.ssl.sslOnly", "true");
+            setConfigurationProperty("connector.ssl.wantClientAuth", "true");
         }
 
         // set the ssl system properties
diff --git a/qpid/java/test-profiles/JavaExcludes b/qpid/java/test-profiles/JavaExcludes
index 9741eed2e9..738467c60f 100644
--- a/qpid/java/test-profiles/JavaExcludes
+++ b/qpid/java/test-profiles/JavaExcludes
@@ -28,9 +28,6 @@ org.apache.qpid.test.client.queue.QueuePolicyTest#testRejectPolicy
 //Moved from JavaStandaloneExcludes when it was removed
 ///////////////////////////////////////////////////////
 
-//The Java broker doesnt support client auth
-org.apache.qpid.client.ssl.SSLTest#testMultipleCertsInSingleStore
-
 //QPID-3605 Durable subscriber with no-local true receives messages on re-connection
 org.apache.qpid.test.unit.topic.DurableSubscriptionTest#testNoLocalMessagesNotDeliveredAfterReconnection
 
diff --git a/qpid/java/test-profiles/JavaPre010Excludes b/qpid/java/test-profiles/JavaPre010Excludes
index 363804abcb..3f29dee203 100644
--- a/qpid/java/test-profiles/JavaPre010Excludes
+++ b/qpid/java/test-profiles/JavaPre010Excludes
@@ -37,6 +37,9 @@ org.apache.qpid.server.queue.AddressBasedSortedQueueTest#*
 org.apache.qpid.test.unit.message.UTF8Test#*
 org.apache.qpid.client.SynchReceiveTest#testReceiveNoWait
 
+// Makes explicit use of 0-10 connection object
+org.apache.qpid.client.ssl.SSLTest#testMultipleCertsInSingleStore
+
 // Tests 0.10 client feature
 org.apache.qpid.test.unit.client.connection.ConnectionTest#testUnsupportedSASLMechanism